diff options
| -rw-r--r-- | tests/test_couchdb.py | 281 |
1 files changed, 271 insertions, 10 deletions
diff --git a/tests/test_couchdb.py b/tests/test_couchdb.py index 58285086..4468ae04 100644 --- a/tests/test_couchdb.py +++ b/tests/test_couchdb.py @@ -1,19 +1,280 @@ -import unittest +import unittest2 from soledad.backends.couch import CouchDatabase +from soledad.backends.leap import LeapDocument +from u1db import errors, vectorclock -class CouchTestCase(unittest.TestCase): +try: + import simplejson as json +except ImportError: + import json # noqa + +simple_doc = '{"key": "value"}' +nested_doc = '{"key": "value", "sub": {"doc": "underneath"}}' + +def make_document_for_test(test, doc_id, rev, content, has_conflicts=False): + return LeapDocument(doc_id, rev, content, has_conflicts=has_conflicts) + +class CouchTestCase(unittest2.TestCase): def setUp(self): - self._db = CouchDatabase('http://localhost:5984', 'u1db_tests') + self.db = CouchDatabase('http://localhost:5984', 'u1db_tests') + + def make_document(self, doc_id, doc_rev, content, has_conflicts=False): + return self.make_document_for_test( + self, doc_id, doc_rev, content, has_conflicts) + + def make_document_for_test(self, test, doc_id, doc_rev, content, + has_conflicts): + return make_document_for_test( + test, doc_id, doc_rev, content, has_conflicts) + + def assertGetDoc(self, db, doc_id, doc_rev, content, has_conflicts): + """Assert that the document in the database looks correct.""" + exp_doc = self.make_document(doc_id, doc_rev, content, + has_conflicts=has_conflicts) + self.assertEqual(exp_doc, db.get_doc(doc_id)) + + def assertGetDocIncludeDeleted(self, db, doc_id, doc_rev, content, + has_conflicts): + """Assert that the document in the database looks correct.""" + exp_doc = self.make_document(doc_id, doc_rev, content, + has_conflicts=has_conflicts) + self.assertEqual(exp_doc, db.get_doc(doc_id, include_deleted=True)) + + + def test_create_doc_allocating_doc_id(self): + doc = self.db.create_doc_from_json(simple_doc) + self.assertNotEqual(None, doc.doc_id) + self.assertNotEqual(None, doc.rev) + self.assertGetDoc(self.db, doc.doc_id, doc.rev, simple_doc, False) + + def test_create_doc_different_ids_same_db(self): + doc1 = self.db.create_doc_from_json(simple_doc) + doc2 = self.db.create_doc_from_json(nested_doc) + self.assertNotEqual(doc1.doc_id, doc2.doc_id) + + def test_create_doc_with_id(self): + doc = self.db.create_doc_from_json(simple_doc, doc_id='my-id') + self.assertEqual('my-id', doc.doc_id) + self.assertNotEqual(None, doc.rev) + self.assertGetDoc(self.db, doc.doc_id, doc.rev, simple_doc, False) + + def test_create_doc_existing_id(self): + doc = self.db.create_doc_from_json(simple_doc) + new_content = '{"something": "else"}' + self.assertRaises( + errors.RevisionConflict, self.db.create_doc_from_json, + new_content, doc.doc_id) + self.assertGetDoc(self.db, doc.doc_id, doc.rev, simple_doc, False) + + def test_put_doc_creating_initial(self): + doc = self.make_document('my_doc_id', None, simple_doc) + new_rev = self.db.put_doc(doc) + self.assertIsNot(None, new_rev) + self.assertGetDoc(self.db, 'my_doc_id', new_rev, simple_doc, False) + + def test_put_doc_space_in_id(self): + doc = self.make_document('my doc id', None, simple_doc) + self.assertRaises(errors.InvalidDocId, self.db.put_doc, doc) + + def test_put_doc_update(self): + doc = self.db.create_doc_from_json(simple_doc, doc_id='my_doc_id') + orig_rev = doc.rev + doc.set_json('{"updated": "stuff"}') + new_rev = self.db.put_doc(doc) + self.assertNotEqual(new_rev, orig_rev) + self.assertGetDoc(self.db, 'my_doc_id', new_rev, + '{"updated": "stuff"}', False) + self.assertEqual(doc.rev, new_rev) + + def test_put_non_ascii_key(self): + content = json.dumps({u'key\xe5': u'val'}) + doc = self.db.create_doc_from_json(content, doc_id='my_doc') + self.assertGetDoc(self.db, 'my_doc', doc.rev, content, False) + + def test_put_non_ascii_value(self): + content = json.dumps({'key': u'\xe5'}) + doc = self.db.create_doc_from_json(content, doc_id='my_doc') + self.assertGetDoc(self.db, 'my_doc', doc.rev, content, False) + + def test_put_doc_refuses_no_id(self): + doc = self.make_document(None, None, simple_doc) + self.assertRaises(errors.InvalidDocId, self.db.put_doc, doc) + doc = self.make_document("", None, simple_doc) + self.assertRaises(errors.InvalidDocId, self.db.put_doc, doc) + + def test_put_doc_refuses_slashes(self): + doc = self.make_document('a/b', None, simple_doc) + self.assertRaises(errors.InvalidDocId, self.db.put_doc, doc) + doc = self.make_document(r'\b', None, simple_doc) + self.assertRaises(errors.InvalidDocId, self.db.put_doc, doc) + + def test_put_doc_url_quoting_is_fine(self): + doc_id = "%2F%2Ffoo%2Fbar" + doc = self.make_document(doc_id, None, simple_doc) + new_rev = self.db.put_doc(doc) + self.assertGetDoc(self.db, doc_id, new_rev, simple_doc, False) + + def test_put_doc_refuses_non_existing_old_rev(self): + doc = self.make_document('doc-id', 'test:4', simple_doc) + self.assertRaises(errors.RevisionConflict, self.db.put_doc, doc) + + def test_put_doc_refuses_non_ascii_doc_id(self): + doc = self.make_document('d\xc3\xa5c-id', None, simple_doc) + self.assertRaises(errors.InvalidDocId, self.db.put_doc, doc) + + def test_put_fails_with_bad_old_rev(self): + doc = self.db.create_doc_from_json(simple_doc, doc_id='my_doc_id') + old_rev = doc.rev + bad_doc = self.make_document(doc.doc_id, 'other:1', + '{"something": "else"}') + self.assertRaises(errors.RevisionConflict, self.db.put_doc, bad_doc) + self.assertGetDoc(self.db, 'my_doc_id', old_rev, simple_doc, False) + + def test_create_succeeds_after_delete(self): + doc = self.db.create_doc_from_json(simple_doc, doc_id='my_doc_id') + self.db.delete_doc(doc) + deleted_doc = self.db.get_doc('my_doc_id', include_deleted=True) + deleted_vc = vectorclock.VectorClockRev(deleted_doc.rev) + new_doc = self.db.create_doc_from_json(simple_doc, doc_id='my_doc_id') + self.assertGetDoc(self.db, 'my_doc_id', new_doc.rev, simple_doc, False) + new_vc = vectorclock.VectorClockRev(new_doc.rev) + self.assertTrue( + new_vc.is_newer(deleted_vc), + "%s does not supersede %s" % (new_doc.rev, deleted_doc.rev)) + + def test_put_succeeds_after_delete(self): + doc = self.db.create_doc_from_json(simple_doc, doc_id='my_doc_id') + self.db.delete_doc(doc) + deleted_doc = self.db.get_doc('my_doc_id', include_deleted=True) + deleted_vc = vectorclock.VectorClockRev(deleted_doc.rev) + doc2 = self.make_document('my_doc_id', None, simple_doc) + self.db.put_doc(doc2) + self.assertGetDoc(self.db, 'my_doc_id', doc2.rev, simple_doc, False) + new_vc = vectorclock.VectorClockRev(doc2.rev) + self.assertTrue( + new_vc.is_newer(deleted_vc), + "%s does not supersede %s" % (doc2.rev, deleted_doc.rev)) + + def test_get_doc_after_put(self): + doc = self.db.create_doc_from_json(simple_doc, doc_id='my_doc_id') + self.assertGetDoc(self.db, 'my_doc_id', doc.rev, simple_doc, False) + + def test_get_doc_nonexisting(self): + self.assertIs(None, self.db.get_doc('non-existing')) + + def test_get_doc_deleted(self): + doc = self.db.create_doc_from_json(simple_doc, doc_id='my_doc_id') + self.db.delete_doc(doc) + self.assertIs(None, self.db.get_doc('my_doc_id')) + + def test_get_doc_include_deleted(self): + doc = self.db.create_doc_from_json(simple_doc, doc_id='my_doc_id') + self.db.delete_doc(doc) + self.assertGetDocIncludeDeleted( + self.db, doc.doc_id, doc.rev, None, False) + + def test_get_docs(self): + doc1 = self.db.create_doc_from_json(simple_doc) + doc2 = self.db.create_doc_from_json(nested_doc) + self.assertEqual([doc1, doc2], + list(self.db.get_docs([doc1.doc_id, doc2.doc_id]))) + + def test_get_docs_deleted(self): + doc1 = self.db.create_doc_from_json(simple_doc) + doc2 = self.db.create_doc_from_json(nested_doc) + self.db.delete_doc(doc1) + self.assertEqual([doc2], + list(self.db.get_docs([doc1.doc_id, doc2.doc_id]))) + + def test_get_docs_include_deleted(self): + doc1 = self.db.create_doc_from_json(simple_doc) + doc2 = self.db.create_doc_from_json(nested_doc) + self.db.delete_doc(doc1) + self.assertEqual( + [doc1, doc2], + list(self.db.get_docs([doc1.doc_id, doc2.doc_id], + include_deleted=True))) + + def test_get_docs_request_ordered(self): + doc1 = self.db.create_doc_from_json(simple_doc) + doc2 = self.db.create_doc_from_json(nested_doc) + self.assertEqual([doc1, doc2], + list(self.db.get_docs([doc1.doc_id, doc2.doc_id]))) + self.assertEqual([doc2, doc1], + list(self.db.get_docs([doc2.doc_id, doc1.doc_id]))) + + def test_get_docs_empty_list(self): + self.assertEqual([], list(self.db.get_docs([]))) + + def test_handles_nested_content(self): + doc = self.db.create_doc_from_json(nested_doc) + self.assertGetDoc(self.db, doc.doc_id, doc.rev, nested_doc, False) + + def test_handles_doc_with_null(self): + doc = self.db.create_doc_from_json('{"key": null}') + self.assertGetDoc(self.db, doc.doc_id, doc.rev, '{"key": null}', False) + + def test_delete_doc(self): + doc = self.db.create_doc_from_json(simple_doc) + self.assertGetDoc(self.db, doc.doc_id, doc.rev, simple_doc, False) + orig_rev = doc.rev + self.db.delete_doc(doc) + self.assertNotEqual(orig_rev, doc.rev) + self.assertGetDocIncludeDeleted( + self.db, doc.doc_id, doc.rev, None, False) + self.assertIs(None, self.db.get_doc(doc.doc_id)) + + def test_delete_doc_non_existent(self): + doc = self.make_document('non-existing', 'other:1', simple_doc) + self.assertRaises(errors.DocumentDoesNotExist, self.db.delete_doc, doc) + + def test_delete_doc_already_deleted(self): + doc = self.db.create_doc_from_json(simple_doc) + self.db.delete_doc(doc) + self.assertRaises(errors.DocumentAlreadyDeleted, + self.db.delete_doc, doc) + self.assertGetDocIncludeDeleted( + self.db, doc.doc_id, doc.rev, None, False) + + def test_delete_doc_bad_rev(self): + doc1 = self.db.create_doc_from_json(simple_doc) + self.assertGetDoc(self.db, doc1.doc_id, doc1.rev, simple_doc, False) + doc2 = self.make_document(doc1.doc_id, 'other:1', simple_doc) + self.assertRaises(errors.RevisionConflict, self.db.delete_doc, doc2) + self.assertGetDoc(self.db, doc1.doc_id, doc1.rev, simple_doc, False) + + def test_delete_doc_sets_content_to_None(self): + doc = self.db.create_doc_from_json(simple_doc) + self.db.delete_doc(doc) + self.assertIs(None, doc.get_json()) + + def test_delete_doc_rev_supersedes(self): + doc = self.db.create_doc_from_json(simple_doc) + doc.set_json(nested_doc) + self.db.put_doc(doc) + doc.set_json('{"fishy": "content"}') + self.db.put_doc(doc) + old_rev = doc.rev + self.db.delete_doc(doc) + cur_vc = vectorclock.VectorClockRev(old_rev) + deleted_vc = vectorclock.VectorClockRev(doc.rev) + self.assertTrue(deleted_vc.is_newer(cur_vc), + "%s does not supersede %s" % (doc.rev, old_rev)) + + def test_delete_then_put(self): + doc = self.db.create_doc_from_json(simple_doc) + self.db.delete_doc(doc) + self.assertGetDocIncludeDeleted( + self.db, doc.doc_id, doc.rev, None, False) + doc.set_json(nested_doc) + self.db.put_doc(doc) + self.assertGetDoc(self.db, doc.doc_id, doc.rev, nested_doc, False) + - def test_create_get(self): - doc1 = self._db.create_doc({"key": "value"}, doc_id="testdoc") - doc2 = self._db.get_doc('testdoc') - self.assertEqual(doc1, doc2, 'error storing/retrieving document.') - self.assertEqual(self._db._get_generation(), 1) def tearDown(self): - self._db._server.delete('u1db_tests') + self.db._server.delete('u1db_tests') if __name__ == '__main__': - unittest.main() + unittest2.main() |