summaryrefslogtreecommitdiff
path: root/common/src/leap/soledad/common/tests
diff options
context:
space:
mode:
Diffstat (limited to 'common/src/leap/soledad/common/tests')
-rw-r--r--common/src/leap/soledad/common/tests/test_couch.py53
-rw-r--r--common/src/leap/soledad/common/tests/test_crypto.py23
-rw-r--r--common/src/leap/soledad/common/tests/test_server.py3
3 files changed, 41 insertions, 38 deletions
diff --git a/common/src/leap/soledad/common/tests/test_couch.py b/common/src/leap/soledad/common/tests/test_couch.py
index 2fb799b7..42edf9fe 100644
--- a/common/src/leap/soledad/common/tests/test_couch.py
+++ b/common/src/leap/soledad/common/tests/test_couch.py
@@ -178,7 +178,12 @@ def copy_couch_database_for_test(test, db):
new_db._conflicts = copy.deepcopy(db._conflicts)
new_db._other_generations = copy.deepcopy(db._other_generations)
new_db._indexes = copy.deepcopy(db._indexes)
- new_db._store_u1db_data()
+ # save u1db data on couch
+ for key in new_db.U1DB_DATA_KEYS:
+ doc_id = '%s%s' % (new_db.U1DB_DATA_DOC_ID_PREFIX, key)
+ doc = new_db._get_doc(doc_id)
+ doc.content = {'content': getattr(new_db, key)}
+ new_db._put_doc(doc)
return new_db
@@ -271,6 +276,14 @@ class CouchDatabaseSyncTargetTests(test_sync.DatabaseSyncTargetTests,
scenarios = (tests.multiply_scenarios(COUCH_SCENARIOS, target_scenarios))
+ def setUp(self):
+ # we implement parents' setUp methods here to prevent from launching
+ # more couch instances then needed.
+ tests.TestCase.setUp(self)
+ self.server = self.server_thread = None
+ self.db, self.st = self.create_db_and_target(self)
+ self.other_changes = []
+
def tearDown(self):
self.db.delete_database()
test_sync.DatabaseSyncTargetTests.tearDown(self)
@@ -345,20 +358,18 @@ class CouchDatabaseStorageTests(CouchDBTestCase):
return [self._listify(i) for i in l]
return l
- def _fetch_u1db_data(self, db):
- cdoc = db._database.get(db.U1DB_DATA_DOC_ID)
- jsonstr = db._database.get_attachment(cdoc, 'u1db_json').getvalue()
- return json.loads(jsonstr)
+ def _fetch_u1db_data(self, db, key):
+ doc = db._get_doc("%s%s" % (db.U1DB_DATA_DOC_ID_PREFIX, key))
+ return doc.content['content']
def test_transaction_log_storage_after_put(self):
db = couch.CouchDatabase('http://localhost:' + str(self.wrapper.port),
'u1db_tests')
db.create_doc({'simple': 'doc'})
- content = self._fetch_u1db_data(db)
+ content = self._fetch_u1db_data(db, db.U1DB_TRANSACTION_LOG_KEY)
self.assertEqual(
self._listify(db._transaction_log),
- self._listify(
- json.loads(b64decode(content[db.U1DB_TRANSACTION_LOG_KEY]))))
+ self._listify(content))
def test_conflict_log_storage_after_put_if_newer(self):
db = couch.CouchDatabase('http://localhost:' + str(self.wrapper.port),
@@ -367,29 +378,27 @@ class CouchDatabaseStorageTests(CouchDBTestCase):
doc.set_json(nested_doc)
doc.rev = db._replica_uid + ':2'
db._force_doc_sync_conflict(doc)
- content = self._fetch_u1db_data(db)
+ content = self._fetch_u1db_data(db, db.U1DB_CONFLICTS_KEY)
self.assertEqual(
self._listify(db._conflicts),
- self._listify(
- json.loads(b64decode(content[db.U1DB_CONFLICTS_KEY]))))
+ self._listify(content))
def test_other_gens_storage_after_set(self):
db = couch.CouchDatabase('http://localhost:' + str(self.wrapper.port),
'u1db_tests')
doc = db.create_doc({'simple': 'doc'})
db._set_replica_gen_and_trans_id('a', 'b', 'c')
- content = self._fetch_u1db_data(db)
+ content = self._fetch_u1db_data(db, db.U1DB_OTHER_GENERATIONS_KEY)
self.assertEqual(
self._listify(db._other_generations),
- self._listify(
- json.loads(b64decode(content[db.U1DB_OTHER_GENERATIONS_KEY]))))
+ self._listify(content))
def test_index_storage_after_create(self):
db = couch.CouchDatabase('http://localhost:' + str(self.wrapper.port),
'u1db_tests')
doc = db.create_doc({'name': 'john'})
db.create_index('myindex', 'name')
- content = self._fetch_u1db_data(db)
+ content = self._fetch_u1db_data(db, db.U1DB_INDEXES_KEY)
myind = db._indexes['myindex']
index = {
'myindex': {
@@ -400,8 +409,7 @@ class CouchDatabaseStorageTests(CouchDBTestCase):
}
self.assertEqual(
self._listify(index),
- self._listify(
- json.loads(b64decode(content[db.U1DB_INDEXES_KEY]))))
+ self._listify(content))
def test_index_storage_after_delete(self):
db = couch.CouchDatabase('http://localhost:' + str(self.wrapper.port),
@@ -410,7 +418,7 @@ class CouchDatabaseStorageTests(CouchDBTestCase):
db.create_index('myindex', 'name')
db.create_index('myindex2', 'name')
db.delete_index('myindex')
- content = self._fetch_u1db_data(db)
+ content = self._fetch_u1db_data(db, db.U1DB_INDEXES_KEY)
myind = db._indexes['myindex2']
index = {
'myindex2': {
@@ -421,16 +429,13 @@ class CouchDatabaseStorageTests(CouchDBTestCase):
}
self.assertEqual(
self._listify(index),
- self._listify(
- json.loads(b64decode(content[db.U1DB_INDEXES_KEY]))))
+ self._listify(content))
def test_replica_uid_storage_after_db_creation(self):
db = couch.CouchDatabase('http://localhost:' + str(self.wrapper.port),
'u1db_tests')
- content = self._fetch_u1db_data(db)
- self.assertEqual(
- db._replica_uid,
- b64decode(content[db.U1DB_REPLICA_UID_KEY]))
+ content = self._fetch_u1db_data(db, db.U1DB_REPLICA_UID_KEY)
+ self.assertEqual(db._replica_uid, content)
load_tests = tests.load_with_scenarios
diff --git a/common/src/leap/soledad/common/tests/test_crypto.py b/common/src/leap/soledad/common/tests/test_crypto.py
index 01b43299..db217bb3 100644
--- a/common/src/leap/soledad/common/tests/test_crypto.py
+++ b/common/src/leap/soledad/common/tests/test_crypto.py
@@ -29,9 +29,6 @@ import binascii
from leap.common.testing.basetest import BaseLeapTest
-from Crypto import Random
-
-
from leap.soledad.client import (
Soledad,
crypto,
@@ -193,7 +190,7 @@ class SoledadCryptoAESTestCase(BaseSoledadTest):
def test_encrypt_decrypt_sym(self):
# generate 256-bit key
- key = Random.new().read(32)
+ key = os.urandom(32)
iv, cyphertext = self._soledad._crypto.encrypt_sym(
'data', key,
method=crypto.EncryptionMethods.AES_256_CTR)
@@ -206,7 +203,7 @@ class SoledadCryptoAESTestCase(BaseSoledadTest):
self.assertEqual('data', plaintext)
def test_decrypt_with_wrong_iv_fails(self):
- key = Random.new().read(32)
+ key = os.urandom(32)
iv, cyphertext = self._soledad._crypto.encrypt_sym(
'data', key,
method=crypto.EncryptionMethods.AES_256_CTR)
@@ -224,17 +221,17 @@ class SoledadCryptoAESTestCase(BaseSoledadTest):
self.assertNotEqual('data', plaintext)
def test_decrypt_with_wrong_key_fails(self):
- key = Random.new().read(32)
+ key = os.urandom(32)
iv, cyphertext = self._soledad._crypto.encrypt_sym(
'data', key,
method=crypto.EncryptionMethods.AES_256_CTR)
self.assertTrue(cyphertext is not None)
self.assertTrue(cyphertext != '')
self.assertTrue(cyphertext != 'data')
- wrongkey = Random.new().read(32) # 256-bits key
+ wrongkey = os.urandom(32) # 256-bits key
# ensure keys are different in case we are extremely lucky
while wrongkey == key:
- wrongkey = Random.new().read(32)
+ wrongkey = os.urandom(32)
plaintext = self._soledad._crypto.decrypt_sym(
cyphertext, wrongkey, iv=iv,
method=crypto.EncryptionMethods.AES_256_CTR)
@@ -245,7 +242,7 @@ class SoledadCryptoXSalsa20TestCase(BaseSoledadTest):
def test_encrypt_decrypt_sym(self):
# generate 256-bit key
- key = Random.new().read(32)
+ key = os.urandom(32)
iv, cyphertext = self._soledad._crypto.encrypt_sym(
'data', key,
method=crypto.EncryptionMethods.XSALSA20)
@@ -258,7 +255,7 @@ class SoledadCryptoXSalsa20TestCase(BaseSoledadTest):
self.assertEqual('data', plaintext)
def test_decrypt_with_wrong_iv_fails(self):
- key = Random.new().read(32)
+ key = os.urandom(32)
iv, cyphertext = self._soledad._crypto.encrypt_sym(
'data', key,
method=crypto.EncryptionMethods.XSALSA20)
@@ -276,17 +273,17 @@ class SoledadCryptoXSalsa20TestCase(BaseSoledadTest):
self.assertNotEqual('data', plaintext)
def test_decrypt_with_wrong_key_fails(self):
- key = Random.new().read(32)
+ key = os.urandom(32)
iv, cyphertext = self._soledad._crypto.encrypt_sym(
'data', key,
method=crypto.EncryptionMethods.XSALSA20)
self.assertTrue(cyphertext is not None)
self.assertTrue(cyphertext != '')
self.assertTrue(cyphertext != 'data')
- wrongkey = Random.new().read(32) # 256-bits key
+ wrongkey = os.urandom(32) # 256-bits key
# ensure keys are different in case we are extremely lucky
while wrongkey == key:
- wrongkey = Random.new().read(32)
+ wrongkey = os.urandom(32)
plaintext = self._soledad._crypto.decrypt_sym(
cyphertext, wrongkey, iv=iv,
method=crypto.EncryptionMethods.XSALSA20)
diff --git a/common/src/leap/soledad/common/tests/test_server.py b/common/src/leap/soledad/common/tests/test_server.py
index beb7e04d..1ea4d615 100644
--- a/common/src/leap/soledad/common/tests/test_server.py
+++ b/common/src/leap/soledad/common/tests/test_server.py
@@ -310,7 +310,8 @@ class EncryptedSyncTestCase(
secret_id=secret_id)
def make_app(self):
- self.request_state = CouchServerState(self._couch_url)
+ self.request_state = CouchServerState(
+ self._couch_url, 'shared', 'tokens', 'user-')
return self.make_app_with_state(self.request_state)
def setUp(self):
generated by cgit v1.2.3 (git 2.25.1) at 2024-06-12 11:12:02 +0000