From a59b34f37fb687de77d1a94f41f53a961baad348 Mon Sep 17 00:00:00 2001 From: drebs Date: Tue, 8 Jan 2013 17:25:06 -0200 Subject: SQLCipherBackend passes all relevant u1db tests. --- src/leap/soledad/backends/sqlcipher.py | 2 +- src/leap/soledad/tests/test_sqlcipher.py | 181 +++++++++++++++++++------ src/leap/soledad/tests/u1db_tests/test_open.py | 69 ++++++++++ 3 files changed, 211 insertions(+), 41 deletions(-) create mode 100644 src/leap/soledad/tests/u1db_tests/test_open.py (limited to 'src') diff --git a/src/leap/soledad/backends/sqlcipher.py b/src/leap/soledad/backends/sqlcipher.py index 34434cf5..3b03bc95 100644 --- a/src/leap/soledad/backends/sqlcipher.py +++ b/src/leap/soledad/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/src/leap/soledad/tests/test_sqlcipher.py b/src/leap/soledad/tests/test_sqlcipher.py index f203dc66..c712bc47 100644 --- a/src/leap/soledad/tests/test_sqlcipher.py +++ b/src/leap/soledad/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/src/leap/soledad/tests/u1db_tests/test_open.py b/src/leap/soledad/tests/u1db_tests/test_open.py new file mode 100644 index 00000000..88312402 --- /dev/null +++ b/src/leap/soledad/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 . + +"""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) -- cgit v1.2.3