summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--backends/sqlcipher.py2
-rw-r--r--tests/test_sqlcipher.py181
-rw-r--r--tests/u1db_tests/test_open.py69
3 files changed, 211 insertions, 41 deletions
diff --git a/backends/sqlcipher.py b/backends/sqlcipher.py
index 34434cf5..3b03bc95 100644
--- a/backends/sqlcipher.py
+++ b/backends/sqlcipher.py
@@ -37,7 +37,7 @@ from u1db import (
)
-def open(path, create, password, document_factory=None):
+def open(path, password, create, document_factory=None):
"""Open a database at the given location.
Will raise u1db.errors.DatabaseDoesNotExist if create=False and the
diff --git a/tests/test_sqlcipher.py b/tests/test_sqlcipher.py
index f203dc66..c712bc47 100644
--- a/tests/test_sqlcipher.py
+++ b/tests/test_sqlcipher.py
@@ -7,22 +7,120 @@ import unittest2 as unittest
from sqlite3 import dbapi2
+# u1db stuff.
from u1db import (
errors,
query_parser,
)
+
+# soledad stuff.
from leap.soledad.backends import sqlcipher as sqlite_backend
+from leap.soledad.backends.sqlcipher import open as u1db_open
from leap.soledad.backends.leap_backend import LeapDocument
+
+# u1db tests stuff.
from leap.soledad.tests import u1db_tests as tests
from leap.soledad.tests.u1db_tests.test_sqlite_backend import (
TestSQLiteDatabase,
TestSQLitePartialExpandDatabase,
)
-from leap.soledad.tests.u1db_tests.test_backends import TestAlternativeDocument
-from leap.soledad.tests.u1db_tests.test_backends import AllDatabaseTests
+from leap.soledad.tests.u1db_tests.test_backends import (
+ TestAlternativeDocument,
+ AllDatabaseTests,
+ LocalDatabaseTests,
+ LocalDatabaseValidateGenNTransIdTests,
+ LocalDatabaseValidateSourceGenTests,
+ LocalDatabaseWithConflictsTests,
+ DatabaseIndexTests,
+)
+from leap.soledad.tests.u1db_tests.test_open import (
+ TestU1DBOpen,
+)
PASSWORD = '123456'
+#-----------------------------------------------------------------------------
+# The following tests come from `u1db.tests.test_common_backends`.
+#-----------------------------------------------------------------------------
+
+class TestSQLCipherBackendImpl(tests.TestCase):
+
+ def test__allocate_doc_id(self):
+ db = sqlite_backend.SQLCipherDatabase(':memory:', PASSWORD)
+ doc_id1 = db._allocate_doc_id()
+ self.assertTrue(doc_id1.startswith('D-'))
+ self.assertEqual(34, len(doc_id1))
+ int(doc_id1[len('D-'):], 16)
+ self.assertNotEqual(doc_id1, db._allocate_doc_id())
+
+
+#-----------------------------------------------------------------------------
+# The following tests come from `u1db.tests.test_backends`.
+#-----------------------------------------------------------------------------
+
+def make_sqlcipher_database_for_test(test, replica_uid):
+ db = sqlite_backend.SQLCipherDatabase(':memory:', PASSWORD)
+ db._set_replica_uid(replica_uid)
+ return db
+
+
+def copy_sqlcipher_database_for_test(test, db):
+ # DO NOT COPY OR REUSE THIS CODE OUTSIDE TESTS: COPYING U1DB DATABASES IS
+ # THE WRONG THING TO DO, THE ONLY REASON WE DO SO HERE IS TO TEST THAT WE
+ # CORRECTLY DETECT IT HAPPENING SO THAT WE CAN RAISE ERRORS RATHER THAN
+ # CORRUPT USER DATA. USE SYNC INSTEAD, OR WE WILL SEND NINJA TO YOUR
+ # HOUSE.
+ new_db = sqlite_backend.SQLCipherDatabase(':memory:', PASSWORD)
+ tmpfile = StringIO()
+ for line in db._db_handle.iterdump():
+ if not 'sqlite_sequence' in line: # work around bug in iterdump
+ tmpfile.write('%s\n' % line)
+ tmpfile.seek(0)
+ new_db._db_handle = dbapi2.connect(':memory:')
+ new_db._db_handle.cursor().executescript(tmpfile.read())
+ new_db._db_handle.commit()
+ new_db._set_replica_uid(db._replica_uid)
+ new_db._factory = db._factory
+ return new_db
+
+
+SQLCIPHER_SCENARIOS = [
+ ('sqlcipher', {'make_database_for_test': make_sqlcipher_database_for_test,
+ 'copy_database_for_test': copy_sqlcipher_database_for_test,
+ 'make_document_for_test': tests.make_document_for_test,}),
+ ]
+
+
+class SQLCipherTests(AllDatabaseTests):
+ scenarios = SQLCIPHER_SCENARIOS
+
+
+class SQLCipherDatabaseTests(LocalDatabaseTests):
+ scenarios = SQLCIPHER_SCENARIOS
+
+
+class SQLCipherValidateGenNTransIdTests(LocalDatabaseValidateGenNTransIdTests):
+ scenarios = SQLCIPHER_SCENARIOS
+
+
+class SQLCipherValidateSourceGenTests(LocalDatabaseValidateSourceGenTests):
+ scenarios = SQLCIPHER_SCENARIOS
+
+
+class SQLCipherWithConflictsTests(LocalDatabaseWithConflictsTests):
+ scenarios = SQLCIPHER_SCENARIOS
+
+
+class SQLCipherIndexTests(DatabaseIndexTests):
+ scenarios = SQLCIPHER_SCENARIOS
+
+
+load_tests = tests.load_with_scenarios
+
+
+#-----------------------------------------------------------------------------
+# The following tests come from `u1db.tests.test_sqlite_backend`.
+#-----------------------------------------------------------------------------
class TestSQLCipherDatabase(TestSQLitePartialExpandDatabase):
@@ -114,41 +212,44 @@ class TestSQLCipherDatabase(TestSQLitePartialExpandDatabase):
'index_storage': 'expand referenced encrypted'}, config)
-# The following tests come from u1db test_backends.
-
-def make_sqlcipher_database_for_test(test, replica_uid):
- db = sqlite_backend.SQLCipherDatabase(':memory:', PASSWORD)
- db._set_replica_uid(replica_uid)
- return db
-
-
-def copy_sqlcipher_database_for_test(test, db):
- # DO NOT COPY OR REUSE THIS CODE OUTSIDE TESTS: COPYING U1DB DATABASES IS
- # THE WRONG THING TO DO, THE ONLY REASON WE DO SO HERE IS TO TEST THAT WE
- # CORRECTLY DETECT IT HAPPENING SO THAT WE CAN RAISE ERRORS RATHER THAN
- # CORRUPT USER DATA. USE SYNC INSTEAD, OR WE WILL SEND NINJA TO YOUR
- # HOUSE.
- new_db = sqlite_backend.SQLCipherDatabase(':memory:', PASSWORD)
- tmpfile = StringIO()
- for line in db._db_handle.iterdump():
- if not 'sqlite_sequence' in line: # work around bug in iterdump
- tmpfile.write('%s\n' % line)
- tmpfile.seek(0)
- new_db._db_handle = dbapi2.connect(':memory:')
- new_db._db_handle.cursor().executescript(tmpfile.read())
- new_db._db_handle.commit()
- new_db._set_replica_uid(db._replica_uid)
- new_db._factory = db._factory
- return new_db
-
-
-class SQLCipherTests(AllDatabaseTests):
-
- scenarios = [
- ('sqlcipher', {'make_database_for_test': make_sqlcipher_database_for_test,
- 'copy_database_for_test': copy_sqlcipher_database_for_test,
- 'make_document_for_test': tests.make_document_for_test,}),
- ]
-
-
-load_tests = tests.load_with_scenarios
+#-----------------------------------------------------------------------------
+# The following tests come from `u1db.tests.test_open`.
+#-----------------------------------------------------------------------------
+
+class SQLCipherOpen(TestU1DBOpen):
+
+ def test_open_no_create(self):
+ self.assertRaises(errors.DatabaseDoesNotExist,
+ u1db_open, self.db_path,
+ password=PASSWORD,
+ create=False)
+ self.assertFalse(os.path.exists(self.db_path))
+
+ def test_open_create(self):
+ db = u1db_open(self.db_path, password=PASSWORD, create=True)
+ self.addCleanup(db.close)
+ self.assertTrue(os.path.exists(self.db_path))
+ self.assertIsInstance(db, sqlite_backend.SQLCipherDatabase)
+
+ def test_open_with_factory(self):
+ db = u1db_open(self.db_path, password=PASSWORD, create=True,
+ document_factory=TestAlternativeDocument)
+ self.addCleanup(db.close)
+ self.assertEqual(TestAlternativeDocument, db._factory)
+
+ def test_open_existing(self):
+ db = sqlite_backend.SQLCipherDatabase(self.db_path, PASSWORD)
+ self.addCleanup(db.close)
+ doc = db.create_doc_from_json(tests.simple_doc)
+ # Even though create=True, we shouldn't wipe the db
+ db2 = u1db_open(self.db_path, password=PASSWORD, create=True)
+ self.addCleanup(db2.close)
+ doc2 = db2.get_doc(doc.doc_id)
+ self.assertEqual(doc, doc2)
+
+ def test_open_existing_no_create(self):
+ db = sqlite_backend.SQLCipherDatabase(self.db_path, PASSWORD)
+ self.addCleanup(db.close)
+ db2 = u1db_open(self.db_path, password=PASSWORD, create=False)
+ self.addCleanup(db2.close)
+ self.assertIsInstance(db2, sqlite_backend.SQLCipherDatabase)
diff --git a/tests/u1db_tests/test_open.py b/tests/u1db_tests/test_open.py
new file mode 100644
index 00000000..88312402
--- /dev/null
+++ b/tests/u1db_tests/test_open.py
@@ -0,0 +1,69 @@
+# Copyright 2011 Canonical Ltd.
+#
+# This file is part of u1db.
+#
+# u1db is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License version 3
+# as published by the Free Software Foundation.
+#
+# u1db is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with u1db. If not, see <http://www.gnu.org/licenses/>.
+
+"""Test u1db.open"""
+
+import os
+
+from u1db import (
+ errors,
+ open as u1db_open,
+ )
+from leap.soledad.tests import u1db_tests as tests
+from u1db.backends import sqlite_backend
+from leap.soledad.tests.u1db_tests.test_backends import TestAlternativeDocument
+
+
+class TestU1DBOpen(tests.TestCase):
+
+ def setUp(self):
+ super(TestU1DBOpen, self).setUp()
+ tmpdir = self.createTempDir()
+ self.db_path = tmpdir + '/test.db'
+
+ def test_open_no_create(self):
+ self.assertRaises(errors.DatabaseDoesNotExist,
+ u1db_open, self.db_path, create=False)
+ self.assertFalse(os.path.exists(self.db_path))
+
+ def test_open_create(self):
+ db = u1db_open(self.db_path, create=True)
+ self.addCleanup(db.close)
+ self.assertTrue(os.path.exists(self.db_path))
+ self.assertIsInstance(db, sqlite_backend.SQLiteDatabase)
+
+ def test_open_with_factory(self):
+ db = u1db_open(self.db_path, create=True,
+ document_factory=TestAlternativeDocument)
+ self.addCleanup(db.close)
+ self.assertEqual(TestAlternativeDocument, db._factory)
+
+ def test_open_existing(self):
+ db = sqlite_backend.SQLitePartialExpandDatabase(self.db_path)
+ self.addCleanup(db.close)
+ doc = db.create_doc_from_json(tests.simple_doc)
+ # Even though create=True, we shouldn't wipe the db
+ db2 = u1db_open(self.db_path, create=True)
+ self.addCleanup(db2.close)
+ doc2 = db2.get_doc(doc.doc_id)
+ self.assertEqual(doc, doc2)
+
+ def test_open_existing_no_create(self):
+ db = sqlite_backend.SQLitePartialExpandDatabase(self.db_path)
+ self.addCleanup(db.close)
+ db2 = u1db_open(self.db_path, create=False)
+ self.addCleanup(db2.close)
+ self.assertIsInstance(db2, sqlite_backend.SQLitePartialExpandDatabase)
generated by cgit v1.2.3 (git 2.25.1) at 2024-12-27 06:37:06 +0000