diff options
Diffstat (limited to 'tests/sqlcipher/test_backend.py')
| -rw-r--r-- | tests/sqlcipher/test_backend.py | 677 |
1 files changed, 677 insertions, 0 deletions
diff --git a/tests/sqlcipher/test_backend.py b/tests/sqlcipher/test_backend.py new file mode 100644 index 00000000..68f8f9f2 --- /dev/null +++ b/tests/sqlcipher/test_backend.py @@ -0,0 +1,677 @@ +# -*- coding: utf-8 -*- +# test_sqlcipher.py +# Copyright (C) 2013 LEAP +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program 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 General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. +""" +Test sqlcipher backend internals. +""" +import os +import pytest +import time +import threading +import sys + +# l2db stuff. +from leap.soledad.common.l2db import errors +from leap.soledad.common.l2db import query_parser + +# soledad stuff. +from leap.soledad.common.document import SoledadDocument +from leap.soledad.client._db.sqlite import SQLitePartialExpandDatabase +from leap.soledad.client._db.sqlcipher import SQLCipherDatabase +from leap.soledad.client._db.sqlcipher import SQLCipherOptions +from leap.soledad.client._db.sqlcipher import DatabaseIsNotEncrypted + +# u1db tests stuff. +from test_soledad import u1db_tests as tests +from test_soledad.u1db_tests import test_backends +from test_soledad.u1db_tests import test_open +from test_soledad.util import SQLCIPHER_SCENARIOS +from test_soledad.util import PASSWORD +from test_soledad.util import BaseSoledadTest + +from testscenarios import TestWithScenarios + +if sys.version_info[0] < 3: + from pysqlcipher import dbapi2 +else: + from pysqlcipher3 import dbapi2 + + +def sqlcipher_open(path, passphrase, create=True, document_factory=None): + return SQLCipherDatabase( + SQLCipherOptions(path, passphrase, create=create)) + + +# ----------------------------------------------------------------------------- +# The following tests come from `u1db.tests.test_common_backend`. +# ----------------------------------------------------------------------------- + +class TestSQLCipherBackendImpl(tests.TestCase): + + def test__allocate_doc_id(self): + db = sqlcipher_open(':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()) + db.close() + + +# ----------------------------------------------------------------------------- +# The following tests come from `u1db.tests.test_backends`. +# ----------------------------------------------------------------------------- + +class SQLCipherTests(TestWithScenarios, test_backends.AllDatabaseTests): + scenarios = SQLCIPHER_SCENARIOS + + +class SQLCipherDatabaseTests(TestWithScenarios, + test_backends.LocalDatabaseTests): + scenarios = SQLCIPHER_SCENARIOS + + +class SQLCipherValidateGenNTransIdTests( + TestWithScenarios, + test_backends.LocalDatabaseValidateGenNTransIdTests): + scenarios = SQLCIPHER_SCENARIOS + + +class SQLCipherValidateSourceGenTests( + TestWithScenarios, + test_backends.LocalDatabaseValidateSourceGenTests): + scenarios = SQLCIPHER_SCENARIOS + + +class SQLCipherWithConflictsTests( + TestWithScenarios, + test_backends.LocalDatabaseWithConflictsTests): + scenarios = SQLCIPHER_SCENARIOS + + +class SQLCipherIndexTests( + TestWithScenarios, test_backends.DatabaseIndexTests): + scenarios = SQLCIPHER_SCENARIOS + + +# ----------------------------------------------------------------------------- +# The following tests come from `u1db.tests.test_sqlite_backend`. +# ----------------------------------------------------------------------------- + +@pytest.mark.usefixtures('method_tmpdir') +class TestSQLCipherDatabase(tests.TestCase): + """ + Tests from u1db.tests.test_sqlite_backend.TestSQLiteDatabase. + """ + + def test_atomic_initialize(self): + # This test was modified to ensure that db2.close() is called within + # the thread that created the database. + dbname = os.path.join(self.tempdir, 'atomic.db') + + t2 = None # will be a thread + + class SQLCipherDatabaseTesting(SQLCipherDatabase): + _index_storage_value = "testing" + + def __init__(self, dbname, ntry): + self._try = ntry + self._is_initialized_invocations = 0 + SQLCipherDatabase.__init__( + self, + SQLCipherOptions(dbname, PASSWORD)) + + def _is_initialized(self, c): + res = \ + SQLCipherDatabase._is_initialized(self, c) + if self._try == 1: + self._is_initialized_invocations += 1 + if self._is_initialized_invocations == 2: + t2.start() + # hard to do better and have a generic test + time.sleep(0.05) + return res + + class SecondTry(threading.Thread): + + outcome2 = [] + + def run(self): + try: + db2 = SQLCipherDatabaseTesting(dbname, 2) + except Exception as e: + SecondTry.outcome2.append(e) + else: + SecondTry.outcome2.append(db2) + + t2 = SecondTry() + db1 = SQLCipherDatabaseTesting(dbname, 1) + t2.join() + + self.assertIsInstance(SecondTry.outcome2[0], SQLCipherDatabaseTesting) + self.assertTrue(db1._is_initialized(db1._get_sqlite_handle().cursor())) + db1.close() + + +@pytest.mark.usefixtures('method_tmpdir') +class TestSQLCipherPartialExpandDatabase(tests.TestCase): + """ + Tests from u1db.tests.test_sqlite_backend.TestSQLitePartialExpandDatabase. + """ + + # The following tests had to be cloned from u1db because they all + # instantiate the backend directly, so we need to change that in order to + # our backend be instantiated in place. + + def setUp(self): + self.db = sqlcipher_open(':memory:', PASSWORD) + + def tearDown(self): + self.db.close() + + def test_default_replica_uid(self): + self.assertIsNot(None, self.db._replica_uid) + self.assertEqual(32, len(self.db._replica_uid)) + int(self.db._replica_uid, 16) + + def test__parse_index(self): + g = self.db._parse_index_definition('fieldname') + self.assertIsInstance(g, query_parser.ExtractField) + self.assertEqual(['fieldname'], g.field) + + def test__update_indexes(self): + g = self.db._parse_index_definition('fieldname') + c = self.db._get_sqlite_handle().cursor() + self.db._update_indexes('doc-id', {'fieldname': 'val'}, + [('fieldname', g)], c) + c.execute('SELECT doc_id, field_name, value FROM document_fields') + self.assertEqual([('doc-id', 'fieldname', 'val')], + c.fetchall()) + + def test_create_database(self): + raw_db = self.db._get_sqlite_handle() + self.assertNotEqual(None, raw_db) + + def test__set_replica_uid(self): + # Start from scratch, so that replica_uid isn't set. + self.assertIsNot(None, self.db._real_replica_uid) + self.assertIsNot(None, self.db._replica_uid) + self.db._set_replica_uid('foo') + c = self.db._get_sqlite_handle().cursor() + c.execute("SELECT value FROM u1db_config WHERE name='replica_uid'") + self.assertEqual(('foo',), c.fetchone()) + self.assertEqual('foo', self.db._real_replica_uid) + self.assertEqual('foo', self.db._replica_uid) + self.db._close_sqlite_handle() + self.assertEqual('foo', self.db._replica_uid) + + def test__open_database(self): + # SQLCipherDatabase has no _open_database() method, so we just pass + # (and test for the same funcionality on test_open_database_existing() + # below). + pass + + def test__open_database_with_factory(self): + # SQLCipherDatabase has no _open_database() method. + pass + + def test__open_database_non_existent(self): + path = self.tempdir + '/non-existent.sqlite' + self.assertRaises(errors.DatabaseDoesNotExist, + sqlcipher_open, + path, PASSWORD, create=False) + + def test__open_database_during_init(self): + # The purpose of this test is to ensure that _open_database() parallel + # db initialization behaviour is correct. As SQLCipherDatabase does + # not have an _open_database() method, we just do not implement this + # test. + pass + + def test__open_database_invalid(self): + # This test was modified to ensure that an empty database file will + # raise a DatabaseIsNotEncrypted exception instead of a + # dbapi2.OperationalError exception. + path1 = self.tempdir + '/invalid1.db' + with open(path1, 'wb') as f: + f.write("") + self.assertRaises(DatabaseIsNotEncrypted, + sqlcipher_open, path1, + PASSWORD) + with open(path1, 'wb') as f: + f.write("invalid") + self.assertRaises(dbapi2.DatabaseError, + sqlcipher_open, path1, + PASSWORD) + + def test_open_database_existing(self): + # In the context of SQLCipherDatabase, where no _open_database() + # method exists and thus there's no call to _which_index_storage(), + # this test tests for the same functionality as + # test_open_database_create() below. So, we just pass. + pass + + def test_open_database_with_factory(self): + # SQLCipherDatabase's constructor has no factory parameter. + pass + + def test_open_database_create(self): + # SQLCipherDatabas has no open_database() method, so we just test for + # the actual database constructor effects. + path = self.tempdir + '/new.sqlite' + db1 = sqlcipher_open(path, PASSWORD, create=True) + db2 = sqlcipher_open(path, PASSWORD, create=False) + self.assertIsInstance(db2, SQLCipherDatabase) + db1.close() + db2.close() + + def test_create_database_initializes_schema(self): + # This test had to be cloned because our implementation of SQLCipher + # backend is referenced with an index_storage_value that includes the + # word "encrypted". See u1db's sqlite_backend and our + # sqlcipher_backend for reference. + raw_db = self.db._get_sqlite_handle() + c = raw_db.cursor() + c.execute("SELECT * FROM u1db_config") + config = dict([(r[0], r[1]) for r in c.fetchall()]) + replica_uid = self.db._replica_uid + self.assertEqual({'sql_schema': '0', 'replica_uid': replica_uid, + 'index_storage': 'expand referenced encrypted'}, + config) + + def test_store_syncable(self): + doc = self.db.create_doc_from_json(tests.simple_doc) + # assert that docs are syncable by default + self.assertEqual(True, doc.syncable) + # assert that we can store syncable = False + doc.syncable = False + self.db.put_doc(doc) + self.assertEqual(False, self.db.get_doc(doc.doc_id).syncable) + # assert that we can store syncable = True + doc.syncable = True + self.db.put_doc(doc) + self.assertEqual(True, self.db.get_doc(doc.doc_id).syncable) + + def test__close_sqlite_handle(self): + raw_db = self.db._get_sqlite_handle() + self.db._close_sqlite_handle() + self.assertRaises(dbapi2.ProgrammingError, + raw_db.cursor) + + def test__get_generation(self): + self.assertEqual(0, self.db._get_generation()) + + def test__get_generation_info(self): + self.assertEqual((0, ''), self.db._get_generation_info()) + + def test_create_index(self): + self.db.create_index('test-idx', "key") + self.assertEqual([('test-idx', ["key"])], self.db.list_indexes()) + + def test_create_index_multiple_fields(self): + self.db.create_index('test-idx', "key", "key2") + self.assertEqual([('test-idx', ["key", "key2"])], + self.db.list_indexes()) + + def test__get_index_definition(self): + self.db.create_index('test-idx', "key", "key2") + # TODO: How would you test that an index is getting used for an SQL + # request? + self.assertEqual(["key", "key2"], + self.db._get_index_definition('test-idx')) + + def test_list_index_mixed(self): + # Make sure that we properly order the output + c = self.db._get_sqlite_handle().cursor() + # We intentionally insert the data in weird ordering, to make sure the + # query still gets it back correctly. + c.executemany("INSERT INTO index_definitions VALUES (?, ?, ?)", + [('idx-1', 0, 'key10'), + ('idx-2', 2, 'key22'), + ('idx-1', 1, 'key11'), + ('idx-2', 0, 'key20'), + ('idx-2', 1, 'key21')]) + self.assertEqual([('idx-1', ['key10', 'key11']), + ('idx-2', ['key20', 'key21', 'key22'])], + self.db.list_indexes()) + + def test_no_indexes_no_document_fields(self): + self.db.create_doc_from_json( + '{"key1": "val1", "key2": "val2"}') + c = self.db._get_sqlite_handle().cursor() + c.execute("SELECT doc_id, field_name, value FROM document_fields" + " ORDER BY doc_id, field_name, value") + self.assertEqual([], c.fetchall()) + + def test_create_extracts_fields(self): + doc1 = self.db.create_doc_from_json('{"key1": "val1", "key2": "val2"}') + doc2 = self.db.create_doc_from_json('{"key1": "valx", "key2": "valy"}') + c = self.db._get_sqlite_handle().cursor() + c.execute("SELECT doc_id, field_name, value FROM document_fields" + " ORDER BY doc_id, field_name, value") + self.assertEqual([], c.fetchall()) + self.db.create_index('test', 'key1', 'key2') + c.execute("SELECT doc_id, field_name, value FROM document_fields" + " ORDER BY doc_id, field_name, value") + self.assertEqual(sorted( + [(doc1.doc_id, "key1", "val1"), + (doc1.doc_id, "key2", "val2"), + (doc2.doc_id, "key1", "valx"), + (doc2.doc_id, "key2", "valy"), ]), sorted(c.fetchall())) + + def test_put_updates_fields(self): + self.db.create_index('test', 'key1', 'key2') + doc1 = self.db.create_doc_from_json( + '{"key1": "val1", "key2": "val2"}') + doc1.content = {"key1": "val1", "key2": "valy"} + self.db.put_doc(doc1) + c = self.db._get_sqlite_handle().cursor() + c.execute("SELECT doc_id, field_name, value FROM document_fields" + " ORDER BY doc_id, field_name, value") + self.assertEqual([(doc1.doc_id, "key1", "val1"), + (doc1.doc_id, "key2", "valy"), ], c.fetchall()) + + def test_put_updates_nested_fields(self): + self.db.create_index('test', 'key', 'sub.doc') + doc1 = self.db.create_doc_from_json(tests.nested_doc) + c = self.db._get_sqlite_handle().cursor() + c.execute("SELECT doc_id, field_name, value FROM document_fields" + " ORDER BY doc_id, field_name, value") + self.assertEqual([(doc1.doc_id, "key", "value"), + (doc1.doc_id, "sub.doc", "underneath"), ], + c.fetchall()) + + def test__ensure_schema_rollback(self): + path = self.tempdir + '/rollback.db' + + class SQLitePartialExpandDbTesting(SQLCipherDatabase): + + def _set_replica_uid_in_transaction(self, uid): + super(SQLitePartialExpandDbTesting, + self)._set_replica_uid_in_transaction(uid) + if fail: + raise Exception() + + db = SQLitePartialExpandDbTesting.__new__(SQLitePartialExpandDbTesting) + db._db_handle = dbapi2.connect(path) # db is there but not yet init-ed + fail = True + self.assertRaises(Exception, db._ensure_schema) + fail = False + db._initialize(db._db_handle.cursor()) + + def test_open_database_non_existent(self): + path = self.tempdir + '/non-existent.sqlite' + self.assertRaises(errors.DatabaseDoesNotExist, + sqlcipher_open, path, "123", + create=False) + + def test_delete_database_existent(self): + path = self.tempdir + '/new.sqlite' + db = sqlcipher_open(path, "123", create=True) + db.close() + SQLCipherDatabase.delete_database(path) + self.assertRaises(errors.DatabaseDoesNotExist, + sqlcipher_open, path, "123", + create=False) + + def test_delete_database_nonexistent(self): + path = self.tempdir + '/non-existent.sqlite' + self.assertRaises(errors.DatabaseDoesNotExist, + SQLCipherDatabase.delete_database, path) + + def test__get_indexed_fields(self): + self.db.create_index('idx1', 'a', 'b') + self.assertEqual(set(['a', 'b']), self.db._get_indexed_fields()) + self.db.create_index('idx2', 'b', 'c') + self.assertEqual(set(['a', 'b', 'c']), self.db._get_indexed_fields()) + + def test_indexed_fields_expanded(self): + self.db.create_index('idx1', 'key1') + doc1 = self.db.create_doc_from_json('{"key1": "val1", "key2": "val2"}') + self.assertEqual(set(['key1']), self.db._get_indexed_fields()) + c = self.db._get_sqlite_handle().cursor() + c.execute("SELECT doc_id, field_name, value FROM document_fields" + " ORDER BY doc_id, field_name, value") + self.assertEqual([(doc1.doc_id, 'key1', 'val1')], c.fetchall()) + + def test_create_index_updates_fields(self): + doc1 = self.db.create_doc_from_json('{"key1": "val1", "key2": "val2"}') + self.db.create_index('idx1', 'key1') + self.assertEqual(set(['key1']), self.db._get_indexed_fields()) + c = self.db._get_sqlite_handle().cursor() + c.execute("SELECT doc_id, field_name, value FROM document_fields" + " ORDER BY doc_id, field_name, value") + self.assertEqual([(doc1.doc_id, 'key1', 'val1')], c.fetchall()) + + def assertFormatQueryEquals(self, exp_statement, exp_args, definition, + values): + statement, args = self.db._format_query(definition, values) + self.assertEqual(exp_statement, statement) + self.assertEqual(exp_args, args) + + def test__format_query(self): + self.assertFormatQueryEquals( + "SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM " + "document d, document_fields d0 LEFT OUTER JOIN conflicts c ON " + "c.doc_id = d.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name " + "= ? AND d0.value = ? GROUP BY d.doc_id, d.doc_rev, d.content " + "ORDER BY d0.value;", ["key1", "a"], + ["key1"], ["a"]) + + def test__format_query2(self): + self.assertFormatQueryEquals( + 'SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM ' + 'document d, document_fields d0, document_fields d1, ' + 'document_fields d2 LEFT OUTER JOIN conflicts c ON c.doc_id = ' + 'd.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name = ? AND ' + 'd0.value = ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND ' + 'd1.value = ? AND d.doc_id = d2.doc_id AND d2.field_name = ? AND ' + 'd2.value = ? GROUP BY d.doc_id, d.doc_rev, d.content ORDER BY ' + 'd0.value, d1.value, d2.value;', + ["key1", "a", "key2", "b", "key3", "c"], + ["key1", "key2", "key3"], ["a", "b", "c"]) + + def test__format_query_wildcard(self): + self.assertFormatQueryEquals( + 'SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM ' + 'document d, document_fields d0, document_fields d1, ' + 'document_fields d2 LEFT OUTER JOIN conflicts c ON c.doc_id = ' + 'd.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name = ? AND ' + 'd0.value = ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND ' + 'd1.value GLOB ? AND d.doc_id = d2.doc_id AND d2.field_name = ? ' + 'AND d2.value NOT NULL GROUP BY d.doc_id, d.doc_rev, d.content ' + 'ORDER BY d0.value, d1.value, d2.value;', + ["key1", "a", "key2", "b*", "key3"], ["key1", "key2", "key3"], + ["a", "b*", "*"]) + + def assertFormatRangeQueryEquals(self, exp_statement, exp_args, definition, + start_value, end_value): + statement, args = self.db._format_range_query( + definition, start_value, end_value) + self.assertEqual(exp_statement, statement) + self.assertEqual(exp_args, args) + + def test__format_range_query(self): + self.assertFormatRangeQueryEquals( + 'SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM ' + 'document d, document_fields d0, document_fields d1, ' + 'document_fields d2 LEFT OUTER JOIN conflicts c ON c.doc_id = ' + 'd.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name = ? AND ' + 'd0.value >= ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND ' + 'd1.value >= ? AND d.doc_id = d2.doc_id AND d2.field_name = ? AND ' + 'd2.value >= ? AND d.doc_id = d0.doc_id AND d0.field_name = ? AND ' + 'd0.value <= ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND ' + 'd1.value <= ? AND d.doc_id = d2.doc_id AND d2.field_name = ? AND ' + 'd2.value <= ? GROUP BY d.doc_id, d.doc_rev, d.content ORDER BY ' + 'd0.value, d1.value, d2.value;', + ['key1', 'a', 'key2', 'b', 'key3', 'c', 'key1', 'p', 'key2', 'q', + 'key3', 'r'], + ["key1", "key2", "key3"], ["a", "b", "c"], ["p", "q", "r"]) + + def test__format_range_query_no_start(self): + self.assertFormatRangeQueryEquals( + 'SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM ' + 'document d, document_fields d0, document_fields d1, ' + 'document_fields d2 LEFT OUTER JOIN conflicts c ON c.doc_id = ' + 'd.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name = ? AND ' + 'd0.value <= ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND ' + 'd1.value <= ? AND d.doc_id = d2.doc_id AND d2.field_name = ? AND ' + 'd2.value <= ? GROUP BY d.doc_id, d.doc_rev, d.content ORDER BY ' + 'd0.value, d1.value, d2.value;', + ['key1', 'a', 'key2', 'b', 'key3', 'c'], + ["key1", "key2", "key3"], None, ["a", "b", "c"]) + + def test__format_range_query_no_end(self): + self.assertFormatRangeQueryEquals( + 'SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM ' + 'document d, document_fields d0, document_fields d1, ' + 'document_fields d2 LEFT OUTER JOIN conflicts c ON c.doc_id = ' + 'd.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name = ? AND ' + 'd0.value >= ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND ' + 'd1.value >= ? AND d.doc_id = d2.doc_id AND d2.field_name = ? AND ' + 'd2.value >= ? GROUP BY d.doc_id, d.doc_rev, d.content ORDER BY ' + 'd0.value, d1.value, d2.value;', + ['key1', 'a', 'key2', 'b', 'key3', 'c'], + ["key1", "key2", "key3"], ["a", "b", "c"], None) + + def test__format_range_query_wildcard(self): + self.assertFormatRangeQueryEquals( + 'SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM ' + 'document d, document_fields d0, document_fields d1, ' + 'document_fields d2 LEFT OUTER JOIN conflicts c ON c.doc_id = ' + 'd.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name = ? AND ' + 'd0.value >= ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND ' + 'd1.value >= ? AND d.doc_id = d2.doc_id AND d2.field_name = ? AND ' + 'd2.value NOT NULL AND d.doc_id = d0.doc_id AND d0.field_name = ? ' + 'AND d0.value <= ? AND d.doc_id = d1.doc_id AND d1.field_name = ? ' + 'AND (d1.value < ? OR d1.value GLOB ?) AND d.doc_id = d2.doc_id ' + 'AND d2.field_name = ? AND d2.value NOT NULL GROUP BY d.doc_id, ' + 'd.doc_rev, d.content ORDER BY d0.value, d1.value, d2.value;', + ['key1', 'a', 'key2', 'b', 'key3', 'key1', 'p', 'key2', 'q', 'q*', + 'key3'], + ["key1", "key2", "key3"], ["a", "b*", "*"], ["p", "q*", "*"]) + + +# ----------------------------------------------------------------------------- +# The following tests come from `u1db.tests.test_open`. +# ----------------------------------------------------------------------------- + + +class SQLCipherOpen(test_open.TestU1DBOpen): + + def test_open_no_create(self): + self.assertRaises(errors.DatabaseDoesNotExist, + sqlcipher_open, self.db_path, + PASSWORD, + create=False) + self.assertFalse(os.path.exists(self.db_path)) + + def test_open_create(self): + db = sqlcipher_open(self.db_path, PASSWORD, create=True) + self.addCleanup(db.close) + self.assertTrue(os.path.exists(self.db_path)) + self.assertIsInstance(db, SQLCipherDatabase) + + def test_open_with_factory(self): + db = sqlcipher_open(self.db_path, PASSWORD, create=True, + document_factory=SoledadDocument) + self.addCleanup(db.close) + doc = db.create_doc({}) + self.assertTrue(isinstance(doc, SoledadDocument)) + + def test_open_existing(self): + db = sqlcipher_open(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 = sqlcipher_open(self.db_path, 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 = sqlcipher_open(self.db_path, PASSWORD) + self.addCleanup(db.close) + db2 = sqlcipher_open(self.db_path, PASSWORD, create=False) + self.addCleanup(db2.close) + self.assertIsInstance(db2, SQLCipherDatabase) + + +# ----------------------------------------------------------------------------- +# Tests for actual encryption of the database +# ----------------------------------------------------------------------------- + +class SQLCipherEncryptionTests(BaseSoledadTest): + + """ + Tests to guarantee SQLCipher is indeed encrypting data when storing. + """ + + def _delete_dbfiles(self): + for dbfile in [self.DB_FILE]: + if os.path.exists(dbfile): + os.unlink(dbfile) + + def setUp(self): + BaseSoledadTest.setUp(self) + self.DB_FILE = os.path.join(self.tempdir, 'test.db') + self._delete_dbfiles() + + def tearDown(self): + self._delete_dbfiles() + BaseSoledadTest.tearDown(self) + + def test_try_to_open_encrypted_db_with_sqlite_backend(self): + """ + SQLite backend should not succeed to open SQLCipher databases. + """ + db = sqlcipher_open(self.DB_FILE, PASSWORD) + doc = db.create_doc_from_json(tests.simple_doc) + db.close() + try: + # trying to open an encrypted database with the regular u1db + # backend should raise a DatabaseError exception. + SQLitePartialExpandDatabase(self.DB_FILE, + document_factory=SoledadDocument) + raise DatabaseIsNotEncrypted() + except dbapi2.DatabaseError: + # at this point we know that the regular U1DB sqlcipher backend + # did not succeed on opening the database, so it was indeed + # encrypted. + db = sqlcipher_open(self.DB_FILE, PASSWORD) + doc = db.get_doc(doc.doc_id) + self.assertEqual(tests.simple_doc, doc.get_json(), + 'decrypted content mismatch') + db.close() + + def test_try_to_open_raw_db_with_sqlcipher_backend(self): + """ + SQLCipher backend should not succeed to open unencrypted databases. + """ + db = SQLitePartialExpandDatabase(self.DB_FILE, + document_factory=SoledadDocument) + db.create_doc_from_json(tests.simple_doc) + db.close() + try: + # trying to open the a non-encrypted database with sqlcipher + # backend should raise a DatabaseIsNotEncrypted exception. + db = sqlcipher_open(self.DB_FILE, PASSWORD) + db.close() + raise dbapi2.DatabaseError( + "SQLCipher backend should not be able to open non-encrypted " + "dbs.") + except DatabaseIsNotEncrypted: + pass |