From 7823990656ac65982a1322ea049298350fb2185e Mon Sep 17 00:00:00 2001 From: drebs Date: Tue, 11 Dec 2012 14:42:57 -0200 Subject: Refactor test files. --- src/leap/soledad/tests/test_sqlcipher.py | 494 +++++++++++++++++++++++++++++++ 1 file changed, 494 insertions(+) create mode 100644 src/leap/soledad/tests/test_sqlcipher.py (limited to 'src/leap/soledad/tests/test_sqlcipher.py') diff --git a/src/leap/soledad/tests/test_sqlcipher.py b/src/leap/soledad/tests/test_sqlcipher.py new file mode 100644 index 00000000..46f27f73 --- /dev/null +++ b/src/leap/soledad/tests/test_sqlcipher.py @@ -0,0 +1,494 @@ +# 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 sqlite backend internals.""" + +import os +import time +import threading + +from sqlite3 import dbapi2 + +from u1db import ( + errors, + tests, + query_parser, + ) +from u1db.backends import sqlite_backend +from u1db.tests.test_backends import TestAlternativeDocument + + +simple_doc = '{"key": "value"}' +nested_doc = '{"key": "value", "sub": {"doc": "underneath"}}' + + +class TestSQLiteDatabase(tests.TestCase): + + def test_atomic_initialize(self): + tmpdir = self.createTempDir() + dbname = os.path.join(tmpdir, 'atomic.db') + + t2 = None # will be a thread + + class SQLiteDatabaseTesting(sqlite_backend.SQLiteDatabase): + _index_storage_value = "testing" + + def __init__(self, dbname, ntry): + self._try = ntry + self._is_initialized_invocations = 0 + super(SQLiteDatabaseTesting, self).__init__(dbname) + + def _is_initialized(self, c): + res = super(SQLiteDatabaseTesting, self)._is_initialized(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 + + outcome2 = [] + + def second_try(): + try: + db2 = SQLiteDatabaseTesting(dbname, 2) + except Exception, e: + outcome2.append(e) + else: + outcome2.append(db2) + + t2 = threading.Thread(target=second_try) + db1 = SQLiteDatabaseTesting(dbname, 1) + t2.join() + + self.assertIsInstance(outcome2[0], SQLiteDatabaseTesting) + db2 = outcome2[0] + self.assertTrue(db2._is_initialized(db1._get_sqlite_handle().cursor())) + + +class TestSQLitePartialExpandDatabase(tests.TestCase): + + def setUp(self): + super(TestSQLitePartialExpandDatabase, self).setUp() + self.db = sqlite_backend.SQLitePartialExpandDatabase(':memory:') + self.db._set_replica_uid('test') + + def test_create_database(self): + raw_db = self.db._get_sqlite_handle() + self.assertNotEqual(None, raw_db) + + def test_default_replica_uid(self): + self.db = sqlite_backend.SQLitePartialExpandDatabase(':memory:') + self.assertIsNot(None, self.db._replica_uid) + self.assertEqual(32, len(self.db._replica_uid)) + int(self.db._replica_uid, 16) + + 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_create_database_initializes_schema(self): + 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()]) + self.assertEqual({'sql_schema': '0', 'replica_uid': 'test', + 'index_storage': 'expand referenced'}, config) + + # These tables must exist, though we don't care what is in them yet + c.execute("SELECT * FROM transaction_log") + c.execute("SELECT * FROM document") + c.execute("SELECT * FROM document_fields") + c.execute("SELECT * FROM sync_log") + c.execute("SELECT * FROM conflicts") + c.execute("SELECT * FROM index_definitions") + + def test__parse_index(self): + self.db = sqlite_backend.SQLitePartialExpandDatabase(':memory:') + g = self.db._parse_index_definition('fieldname') + self.assertIsInstance(g, query_parser.ExtractField) + self.assertEqual(['fieldname'], g.field) + + def test__update_indexes(self): + self.db = sqlite_backend.SQLitePartialExpandDatabase(':memory:') + 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__set_replica_uid(self): + # Start from scratch, so that replica_uid isn't set. + self.db = sqlite_backend.SQLitePartialExpandDatabase(':memory:') + 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__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(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): + temp_dir = self.createTempDir(prefix='u1db-test-') + path = temp_dir + '/rollback.db' + + class SQLitePartialExpandDbTesting( + sqlite_backend.SQLitePartialExpandDatabase): + + 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(self): + temp_dir = self.createTempDir(prefix='u1db-test-') + path = temp_dir + '/test.sqlite' + sqlite_backend.SQLitePartialExpandDatabase(path) + db2 = sqlite_backend.SQLiteDatabase._open_database(path) + self.assertIsInstance(db2, sqlite_backend.SQLitePartialExpandDatabase) + + def test__open_database_with_factory(self): + temp_dir = self.createTempDir(prefix='u1db-test-') + path = temp_dir + '/test.sqlite' + sqlite_backend.SQLitePartialExpandDatabase(path) + db2 = sqlite_backend.SQLiteDatabase._open_database( + path, document_factory=TestAlternativeDocument) + self.assertEqual(TestAlternativeDocument, db2._factory) + + def test__open_database_non_existent(self): + temp_dir = self.createTempDir(prefix='u1db-test-') + path = temp_dir + '/non-existent.sqlite' + self.assertRaises(errors.DatabaseDoesNotExist, + sqlite_backend.SQLiteDatabase._open_database, path) + + def test__open_database_during_init(self): + temp_dir = self.createTempDir(prefix='u1db-test-') + path = temp_dir + '/initialised.db' + db = sqlite_backend.SQLitePartialExpandDatabase.__new__( + sqlite_backend.SQLitePartialExpandDatabase) + db._db_handle = dbapi2.connect(path) # db is there but not yet init-ed + self.addCleanup(db.close) + observed = [] + + class SQLiteDatabaseTesting(sqlite_backend.SQLiteDatabase): + WAIT_FOR_PARALLEL_INIT_HALF_INTERVAL = 0.1 + + @classmethod + def _which_index_storage(cls, c): + res = super(SQLiteDatabaseTesting, cls)._which_index_storage(c) + db._ensure_schema() # init db + observed.append(res[0]) + return res + + db2 = SQLiteDatabaseTesting._open_database(path) + self.addCleanup(db2.close) + self.assertIsInstance(db2, sqlite_backend.SQLitePartialExpandDatabase) + self.assertEqual([None, + sqlite_backend.SQLitePartialExpandDatabase._index_storage_value], + observed) + + def test__open_database_invalid(self): + class SQLiteDatabaseTesting(sqlite_backend.SQLiteDatabase): + WAIT_FOR_PARALLEL_INIT_HALF_INTERVAL = 0.1 + temp_dir = self.createTempDir(prefix='u1db-test-') + path1 = temp_dir + '/invalid1.db' + with open(path1, 'wb') as f: + f.write("") + self.assertRaises(dbapi2.OperationalError, + SQLiteDatabaseTesting._open_database, path1) + with open(path1, 'wb') as f: + f.write("invalid") + self.assertRaises(dbapi2.DatabaseError, + SQLiteDatabaseTesting._open_database, path1) + + def test_open_database_existing(self): + temp_dir = self.createTempDir(prefix='u1db-test-') + path = temp_dir + '/existing.sqlite' + sqlite_backend.SQLitePartialExpandDatabase(path) + db2 = sqlite_backend.SQLiteDatabase.open_database(path, create=False) + self.assertIsInstance(db2, sqlite_backend.SQLitePartialExpandDatabase) + + def test_open_database_with_factory(self): + temp_dir = self.createTempDir(prefix='u1db-test-') + path = temp_dir + '/existing.sqlite' + sqlite_backend.SQLitePartialExpandDatabase(path) + db2 = sqlite_backend.SQLiteDatabase.open_database( + path, create=False, document_factory=TestAlternativeDocument) + self.assertEqual(TestAlternativeDocument, db2._factory) + + def test_open_database_create(self): + temp_dir = self.createTempDir(prefix='u1db-test-') + path = temp_dir + '/new.sqlite' + sqlite_backend.SQLiteDatabase.open_database(path, create=True) + db2 = sqlite_backend.SQLiteDatabase.open_database(path, create=False) + self.assertIsInstance(db2, sqlite_backend.SQLitePartialExpandDatabase) + + def test_open_database_non_existent(self): + temp_dir = self.createTempDir(prefix='u1db-test-') + path = temp_dir + '/non-existent.sqlite' + self.assertRaises(errors.DatabaseDoesNotExist, + sqlite_backend.SQLiteDatabase.open_database, path, + create=False) + + def test_delete_database_existent(self): + temp_dir = self.createTempDir(prefix='u1db-test-') + path = temp_dir + '/new.sqlite' + db = sqlite_backend.SQLiteDatabase.open_database(path, create=True) + db.close() + sqlite_backend.SQLiteDatabase.delete_database(path) + self.assertRaises(errors.DatabaseDoesNotExist, + sqlite_backend.SQLiteDatabase.open_database, path, + create=False) + + def test_delete_database_nonexistent(self): + temp_dir = self.createTempDir(prefix='u1db-test-') + path = temp_dir + '/non-existent.sqlite' + self.assertRaises(errors.DatabaseDoesNotExist, + sqlite_backend.SQLiteDatabase.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*", "*"]) + -- cgit v1.2.3 From a12b80b23695dd1db8ac5edeb4b79e6ff8e527c2 Mon Sep 17 00:00:00 2001 From: drebs Date: Tue, 11 Dec 2012 15:03:12 -0200 Subject: Fix SQLCipherDatabase and add tests. --- src/leap/soledad/tests/test_sqlcipher.py | 84 +++++++++++++++++--------------- 1 file changed, 44 insertions(+), 40 deletions(-) (limited to 'src/leap/soledad/tests/test_sqlcipher.py') diff --git a/src/leap/soledad/tests/test_sqlcipher.py b/src/leap/soledad/tests/test_sqlcipher.py index 46f27f73..e35a6d90 100644 --- a/src/leap/soledad/tests/test_sqlcipher.py +++ b/src/leap/soledad/tests/test_sqlcipher.py @@ -19,16 +19,17 @@ import os import time import threading +import unittest2 as unittest from sqlite3 import dbapi2 from u1db import ( errors, - tests, query_parser, ) -from u1db.backends import sqlite_backend -from u1db.tests.test_backends import TestAlternativeDocument +from soledad.backends import sqlcipher +from soledad.backends.leap import LeapDocument +from soledad import tests simple_doc = '{"key": "value"}' @@ -43,7 +44,7 @@ class TestSQLiteDatabase(tests.TestCase): t2 = None # will be a thread - class SQLiteDatabaseTesting(sqlite_backend.SQLiteDatabase): + class SQLiteDatabaseTesting(sqlcipher.SQLCipherDatabase): _index_storage_value = "testing" def __init__(self, dbname, ntry): @@ -84,7 +85,7 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): def setUp(self): super(TestSQLitePartialExpandDatabase, self).setUp() - self.db = sqlite_backend.SQLitePartialExpandDatabase(':memory:') + self.db = sqlcipher.SQLCipherDatabase(':memory:') self.db._set_replica_uid('test') def test_create_database(self): @@ -92,7 +93,7 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): self.assertNotEqual(None, raw_db) def test_default_replica_uid(self): - self.db = sqlite_backend.SQLitePartialExpandDatabase(':memory:') + self.db = sqlcipher.SQLCipherDatabase(':memory:') self.assertIsNot(None, self.db._replica_uid) self.assertEqual(32, len(self.db._replica_uid)) int(self.db._replica_uid, 16) @@ -109,7 +110,7 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): c.execute("SELECT * FROM u1db_config") config = dict([(r[0], r[1]) for r in c.fetchall()]) self.assertEqual({'sql_schema': '0', 'replica_uid': 'test', - 'index_storage': 'expand referenced'}, config) + 'index_storage': 'expand referenced encrypted'}, config) # These tables must exist, though we don't care what is in them yet c.execute("SELECT * FROM transaction_log") @@ -120,13 +121,13 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): c.execute("SELECT * FROM index_definitions") def test__parse_index(self): - self.db = sqlite_backend.SQLitePartialExpandDatabase(':memory:') + self.db = sqlcipher.SQLCipherDatabase(':memory:') g = self.db._parse_index_definition('fieldname') self.assertIsInstance(g, query_parser.ExtractField) self.assertEqual(['fieldname'], g.field) def test__update_indexes(self): - self.db = sqlite_backend.SQLitePartialExpandDatabase(':memory:') + self.db = sqlcipher.SQLCipherDatabase(':memory:') g = self.db._parse_index_definition('fieldname') c = self.db._get_sqlite_handle().cursor() self.db._update_indexes('doc-id', {'fieldname': 'val'}, @@ -137,7 +138,7 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): def test__set_replica_uid(self): # Start from scratch, so that replica_uid isn't set. - self.db = sqlite_backend.SQLitePartialExpandDatabase(':memory:') + self.db = sqlcipher.SQLCipherDatabase(':memory:') self.assertIsNot(None, self.db._real_replica_uid) self.assertIsNot(None, self.db._replica_uid) self.db._set_replica_uid('foo') @@ -239,7 +240,7 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): path = temp_dir + '/rollback.db' class SQLitePartialExpandDbTesting( - sqlite_backend.SQLitePartialExpandDatabase): + sqlcipher.SQLCipherDatabase): def _set_replica_uid_in_transaction(self, uid): super(SQLitePartialExpandDbTesting, @@ -257,34 +258,34 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): def test__open_database(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/test.sqlite' - sqlite_backend.SQLitePartialExpandDatabase(path) - db2 = sqlite_backend.SQLiteDatabase._open_database(path) - self.assertIsInstance(db2, sqlite_backend.SQLitePartialExpandDatabase) + sqlcipher.SQLCipherDatabase(path) + db2 = sqlcipher.SQLCipherDatabase._open_database(path) + self.assertIsInstance(db2, sqlcipher.SQLCipherDatabase) def test__open_database_with_factory(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/test.sqlite' - sqlite_backend.SQLitePartialExpandDatabase(path) - db2 = sqlite_backend.SQLiteDatabase._open_database( - path, document_factory=TestAlternativeDocument) - self.assertEqual(TestAlternativeDocument, db2._factory) + sqlcipher.SQLCipherDatabase(path) + db2 = sqlcipher.SQLCipherDatabase._open_database( + path, document_factory=LeapDocument) + self.assertEqual(LeapDocument, db2._factory) def test__open_database_non_existent(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/non-existent.sqlite' self.assertRaises(errors.DatabaseDoesNotExist, - sqlite_backend.SQLiteDatabase._open_database, path) + sqlcipher.SQLCipherDatabase._open_database, path) def test__open_database_during_init(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/initialised.db' - db = sqlite_backend.SQLitePartialExpandDatabase.__new__( - sqlite_backend.SQLitePartialExpandDatabase) + db = sqlcipher.SQLCipherDatabase.__new__( + sqlcipher.SQLCipherDatabase) db._db_handle = dbapi2.connect(path) # db is there but not yet init-ed self.addCleanup(db.close) observed = [] - class SQLiteDatabaseTesting(sqlite_backend.SQLiteDatabase): + class SQLiteDatabaseTesting(sqlcipher.SQLCipherDatabase): WAIT_FOR_PARALLEL_INIT_HALF_INTERVAL = 0.1 @classmethod @@ -296,13 +297,13 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): db2 = SQLiteDatabaseTesting._open_database(path) self.addCleanup(db2.close) - self.assertIsInstance(db2, sqlite_backend.SQLitePartialExpandDatabase) + self.assertIsInstance(db2, sqlcipher.SQLCipherDatabase) self.assertEqual([None, - sqlite_backend.SQLitePartialExpandDatabase._index_storage_value], + sqlcipher.SQLCipherDatabase._index_storage_value], observed) def test__open_database_invalid(self): - class SQLiteDatabaseTesting(sqlite_backend.SQLiteDatabase): + class SQLiteDatabaseTesting(sqlcipher.SQLCipherDatabase): WAIT_FOR_PARALLEL_INIT_HALF_INTERVAL = 0.1 temp_dir = self.createTempDir(prefix='u1db-test-') path1 = temp_dir + '/invalid1.db' @@ -318,47 +319,47 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): def test_open_database_existing(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/existing.sqlite' - sqlite_backend.SQLitePartialExpandDatabase(path) - db2 = sqlite_backend.SQLiteDatabase.open_database(path, create=False) - self.assertIsInstance(db2, sqlite_backend.SQLitePartialExpandDatabase) + sqlcipher.SQLCipherDatabase(path) + db2 = sqlcipher.SQLCipherDatabase.open_database(path, create=False) + self.assertIsInstance(db2, sqlcipher.SQLCipherDatabase) def test_open_database_with_factory(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/existing.sqlite' - sqlite_backend.SQLitePartialExpandDatabase(path) - db2 = sqlite_backend.SQLiteDatabase.open_database( - path, create=False, document_factory=TestAlternativeDocument) - self.assertEqual(TestAlternativeDocument, db2._factory) + sqlcipher.SQLCipherDatabase(path) + db2 = sqlcipher.SQLCipherDatabase.open_database( + path, create=False, document_factory=LeapDocument) + self.assertEqual(LeapDocument, db2._factory) def test_open_database_create(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/new.sqlite' - sqlite_backend.SQLiteDatabase.open_database(path, create=True) - db2 = sqlite_backend.SQLiteDatabase.open_database(path, create=False) - self.assertIsInstance(db2, sqlite_backend.SQLitePartialExpandDatabase) + sqlcipher.SQLCipherDatabase.open_database(path, create=True) + db2 = sqlcipher.SQLCipherDatabase.open_database(path, create=False) + self.assertIsInstance(db2, sqlcipher.SQLCipherDatabase) def test_open_database_non_existent(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/non-existent.sqlite' self.assertRaises(errors.DatabaseDoesNotExist, - sqlite_backend.SQLiteDatabase.open_database, path, + sqlcipher.SQLCipherDatabase.open_database, path, create=False) def test_delete_database_existent(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/new.sqlite' - db = sqlite_backend.SQLiteDatabase.open_database(path, create=True) + db = sqlcipher.SQLCipherDatabase.open_database(path, create=True) db.close() - sqlite_backend.SQLiteDatabase.delete_database(path) + sqlcipher.SQLCipherDatabase.delete_database(path) self.assertRaises(errors.DatabaseDoesNotExist, - sqlite_backend.SQLiteDatabase.open_database, path, + sqlcipher.SQLCipherDatabase.open_database, path, create=False) def test_delete_database_nonexistent(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/non-existent.sqlite' self.assertRaises(errors.DatabaseDoesNotExist, - sqlite_backend.SQLiteDatabase.delete_database, path) + sqlcipher.SQLCipherDatabase.delete_database, path) def test__get_indexed_fields(self): self.db.create_index('idx1', 'a', 'b') @@ -492,3 +493,6 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): 'key3'], ["key1", "key2", "key3"], ["a", "b*", "*"], ["p", "q*", "*"]) + +if __name__ == '__main__': + unittest.main() -- cgit v1.2.3 From ece9f7c2116fa961cafabcc6a5790206412c95ae Mon Sep 17 00:00:00 2001 From: drebs Date: Thu, 13 Dec 2012 13:46:27 -0200 Subject: Enforce password on SQLCipher backend. --- src/leap/soledad/tests/test_sqlcipher.py | 79 +++++++++++++++++--------------- 1 file changed, 42 insertions(+), 37 deletions(-) (limited to 'src/leap/soledad/tests/test_sqlcipher.py') diff --git a/src/leap/soledad/tests/test_sqlcipher.py b/src/leap/soledad/tests/test_sqlcipher.py index e35a6d90..f9e9f681 100644 --- a/src/leap/soledad/tests/test_sqlcipher.py +++ b/src/leap/soledad/tests/test_sqlcipher.py @@ -36,7 +36,7 @@ simple_doc = '{"key": "value"}' nested_doc = '{"key": "value", "sub": {"doc": "underneath"}}' -class TestSQLiteDatabase(tests.TestCase): +class TestSQLCipherDatabase(tests.TestCase): def test_atomic_initialize(self): tmpdir = self.createTempDir() @@ -44,16 +44,17 @@ class TestSQLiteDatabase(tests.TestCase): t2 = None # will be a thread - class SQLiteDatabaseTesting(sqlcipher.SQLCipherDatabase): + class SQLCipherDatabaseTesting(sqlcipher.SQLCipherDatabase): _index_storage_value = "testing" def __init__(self, dbname, ntry): self._try = ntry self._is_initialized_invocations = 0 - super(SQLiteDatabaseTesting, self).__init__(dbname) + password = '123456' + super(SQLCipherDatabaseTesting, self).__init__(dbname, password) def _is_initialized(self, c): - res = super(SQLiteDatabaseTesting, self)._is_initialized(c) + res = super(SQLCipherDatabaseTesting, self)._is_initialized(c) if self._try == 1: self._is_initialized_invocations += 1 if self._is_initialized_invocations == 2: @@ -66,26 +67,29 @@ class TestSQLiteDatabase(tests.TestCase): def second_try(): try: - db2 = SQLiteDatabaseTesting(dbname, 2) + db2 = SQLCipherDatabaseTesting(dbname, 2) except Exception, e: outcome2.append(e) else: outcome2.append(db2) t2 = threading.Thread(target=second_try) - db1 = SQLiteDatabaseTesting(dbname, 1) + db1 = SQLCipherDatabaseTesting(dbname, 1) t2.join() - self.assertIsInstance(outcome2[0], SQLiteDatabaseTesting) + self.assertIsInstance(outcome2[0], SQLCipherDatabaseTesting) db2 = outcome2[0] self.assertTrue(db2._is_initialized(db1._get_sqlite_handle().cursor())) -class TestSQLitePartialExpandDatabase(tests.TestCase): +_password = '123456' + + +class TestSQLCipherPartialExpandDatabase(tests.TestCase): def setUp(self): - super(TestSQLitePartialExpandDatabase, self).setUp() - self.db = sqlcipher.SQLCipherDatabase(':memory:') + super(TestSQLCipherPartialExpandDatabase, self).setUp() + self.db = sqlcipher.SQLCipherDatabase(':memory:', _password) self.db._set_replica_uid('test') def test_create_database(self): @@ -93,7 +97,7 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): self.assertNotEqual(None, raw_db) def test_default_replica_uid(self): - self.db = sqlcipher.SQLCipherDatabase(':memory:') + self.db = sqlcipher.SQLCipherDatabase(':memory:', _password) self.assertIsNot(None, self.db._replica_uid) self.assertEqual(32, len(self.db._replica_uid)) int(self.db._replica_uid, 16) @@ -121,13 +125,13 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): c.execute("SELECT * FROM index_definitions") def test__parse_index(self): - self.db = sqlcipher.SQLCipherDatabase(':memory:') + self.db = sqlcipher.SQLCipherDatabase(':memory:', _password) g = self.db._parse_index_definition('fieldname') self.assertIsInstance(g, query_parser.ExtractField) self.assertEqual(['fieldname'], g.field) def test__update_indexes(self): - self.db = sqlcipher.SQLCipherDatabase(':memory:') + self.db = sqlcipher.SQLCipherDatabase(':memory:', _password) g = self.db._parse_index_definition('fieldname') c = self.db._get_sqlite_handle().cursor() self.db._update_indexes('doc-id', {'fieldname': 'val'}, @@ -138,7 +142,7 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): def test__set_replica_uid(self): # Start from scratch, so that replica_uid isn't set. - self.db = sqlcipher.SQLCipherDatabase(':memory:') + self.db = sqlcipher.SQLCipherDatabase(':memory:', _password) self.assertIsNot(None, self.db._real_replica_uid) self.assertIsNot(None, self.db._replica_uid) self.db._set_replica_uid('foo') @@ -239,16 +243,16 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/rollback.db' - class SQLitePartialExpandDbTesting( + class SQLCipherPartialExpandDbTesting( sqlcipher.SQLCipherDatabase): def _set_replica_uid_in_transaction(self, uid): - super(SQLitePartialExpandDbTesting, + super(SQLCipherPartialExpandDbTesting, self)._set_replica_uid_in_transaction(uid) if fail: raise Exception() - db = SQLitePartialExpandDbTesting.__new__(SQLitePartialExpandDbTesting) + db = SQLCipherPartialExpandDbTesting.__new__(SQLCipherPartialExpandDbTesting) db._db_handle = dbapi2.connect(path) # db is there but not yet init-ed fail = True self.assertRaises(Exception, db._ensure_schema) @@ -258,23 +262,23 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): def test__open_database(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/test.sqlite' - sqlcipher.SQLCipherDatabase(path) - db2 = sqlcipher.SQLCipherDatabase._open_database(path) + sqlcipher.SQLCipherDatabase(path, _password) + db2 = sqlcipher.SQLCipherDatabase._open_database(path, _password) self.assertIsInstance(db2, sqlcipher.SQLCipherDatabase) def test__open_database_with_factory(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/test.sqlite' - sqlcipher.SQLCipherDatabase(path) + sqlcipher.SQLCipherDatabase(path, _password) db2 = sqlcipher.SQLCipherDatabase._open_database( - path, document_factory=LeapDocument) + path, _password, document_factory=LeapDocument) self.assertEqual(LeapDocument, db2._factory) def test__open_database_non_existent(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/non-existent.sqlite' self.assertRaises(errors.DatabaseDoesNotExist, - sqlcipher.SQLCipherDatabase._open_database, path) + sqlcipher.SQLCipherDatabase._open_database, path, _password) def test__open_database_during_init(self): temp_dir = self.createTempDir(prefix='u1db-test-') @@ -285,17 +289,17 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): self.addCleanup(db.close) observed = [] - class SQLiteDatabaseTesting(sqlcipher.SQLCipherDatabase): + class SQLCipherDatabaseTesting(sqlcipher.SQLCipherDatabase): WAIT_FOR_PARALLEL_INIT_HALF_INTERVAL = 0.1 @classmethod def _which_index_storage(cls, c): - res = super(SQLiteDatabaseTesting, cls)._which_index_storage(c) + res = super(SQLCipherDatabaseTesting, cls)._which_index_storage(c) db._ensure_schema() # init db observed.append(res[0]) return res - db2 = SQLiteDatabaseTesting._open_database(path) + db2 = SQLCipherDatabaseTesting._open_database(path, _password) self.addCleanup(db2.close) self.assertIsInstance(db2, sqlcipher.SQLCipherDatabase) self.assertEqual([None, @@ -303,39 +307,40 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): observed) def test__open_database_invalid(self): - class SQLiteDatabaseTesting(sqlcipher.SQLCipherDatabase): + class SQLCipherDatabaseTesting(sqlcipher.SQLCipherDatabase): WAIT_FOR_PARALLEL_INIT_HALF_INTERVAL = 0.1 temp_dir = self.createTempDir(prefix='u1db-test-') path1 = temp_dir + '/invalid1.db' with open(path1, 'wb') as f: f.write("") self.assertRaises(dbapi2.OperationalError, - SQLiteDatabaseTesting._open_database, path1) + SQLCipherDatabaseTesting._open_database, path1, _password) with open(path1, 'wb') as f: f.write("invalid") self.assertRaises(dbapi2.DatabaseError, - SQLiteDatabaseTesting._open_database, path1) + SQLCipherDatabaseTesting._open_database, path1, _password) def test_open_database_existing(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/existing.sqlite' - sqlcipher.SQLCipherDatabase(path) - db2 = sqlcipher.SQLCipherDatabase.open_database(path, create=False) + sqlcipher.SQLCipherDatabase(path, _password) + db2 = sqlcipher.SQLCipherDatabase.open_database(path, _password, + create=False) self.assertIsInstance(db2, sqlcipher.SQLCipherDatabase) def test_open_database_with_factory(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/existing.sqlite' - sqlcipher.SQLCipherDatabase(path) + sqlcipher.SQLCipherDatabase(path, _password) db2 = sqlcipher.SQLCipherDatabase.open_database( - path, create=False, document_factory=LeapDocument) + path, _password, create=False, document_factory=LeapDocument) self.assertEqual(LeapDocument, db2._factory) def test_open_database_create(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/new.sqlite' - sqlcipher.SQLCipherDatabase.open_database(path, create=True) - db2 = sqlcipher.SQLCipherDatabase.open_database(path, create=False) + sqlcipher.SQLCipherDatabase.open_database(path, _password, create=True) + db2 = sqlcipher.SQLCipherDatabase.open_database(path, _password, create=False) self.assertIsInstance(db2, sqlcipher.SQLCipherDatabase) def test_open_database_non_existent(self): @@ -343,17 +348,17 @@ class TestSQLitePartialExpandDatabase(tests.TestCase): path = temp_dir + '/non-existent.sqlite' self.assertRaises(errors.DatabaseDoesNotExist, sqlcipher.SQLCipherDatabase.open_database, path, - create=False) + _password, create=False) def test_delete_database_existent(self): temp_dir = self.createTempDir(prefix='u1db-test-') path = temp_dir + '/new.sqlite' - db = sqlcipher.SQLCipherDatabase.open_database(path, create=True) + db = sqlcipher.SQLCipherDatabase.open_database(path, _password, create=True) db.close() sqlcipher.SQLCipherDatabase.delete_database(path) self.assertRaises(errors.DatabaseDoesNotExist, sqlcipher.SQLCipherDatabase.open_database, path, - create=False) + _password, create=False) def test_delete_database_nonexistent(self): temp_dir = self.createTempDir(prefix='u1db-test-') -- cgit v1.2.3