[bug] fixes concurrent sync and their tests

Changes threading.lock to DeferredLock and checks syncing attribute by
looking into the lock state.
Also, applies more of startTwistedServer on tests that relies on
HTTP/1.1.
Fixes mock for events

6 files changed, 31 insertions, 69 deletions

diff --git a/client/src/leap/soledad/client/api.py b/client/src/leap/soledad/client/api.py
index 5ba93721..9000029f 100644
--- a/client/src/leap/soledad/client/api.py
+++ b/client/src/leap/soledad/client/api.py
@@ -47,6 +47,7 @@ from zope.interface import implements
 
 from leap.common.config import get_path_prefix
 from leap.common.plugins import collect_plugins
+from twisted.internet import defer
 
 from leap.soledad.common import SHARED_DB_NAME
 from leap.soledad.common import soledad_assert
@@ -199,6 +200,7 @@ class Soledad(object):
 
         self._crypto = SoledadCrypto(self._secrets.remote_storage_secret)
         self._init_u1db_sqlcipher_backend()
+        self.sync_lock = defer.DeferredLock()
 
         if syncable:
             self._init_u1db_syncer()
@@ -669,10 +671,10 @@ class Soledad(object):
         # -----------------------------------------------------------------
 
         sync_url = urlparse.urljoin(self._server_url, 'user-%s' % self.uuid)
-        d = self._dbsyncer.sync(
+        d = self.sync_lock.run(lambda: self._dbsyncer.sync(
             sync_url,
             creds=self._creds,
-            defer_decryption=defer_decryption)
+            defer_decryption=defer_decryption))
 
         def _sync_callback(local_gen):
             self._last_received_docs = docs = self._dbsyncer.received_docs
@@ -711,7 +713,7 @@ class Soledad(object):
         :return: Wether Soledad is currently synchronizing with the server.
         :rtype: bool
         """
-        return self._dbsyncer.syncing
+        return self.sync_lock.locked
 
     def _set_token(self, token):
         """
diff --git a/client/src/leap/soledad/client/sqlcipher.py b/client/src/leap/soledad/client/sqlcipher.py
index a76a35b7..249ffb1a 100644
--- a/client/src/leap/soledad/client/sqlcipher.py
+++ b/client/src/leap/soledad/client/sqlcipher.py
@@ -43,7 +43,6 @@ handled by Soledad should be created by SQLCipher >= 2.0.
 """
 import logging
 import os
-import threading
 import json
 import u1db
 
@@ -51,8 +50,6 @@ from u1db import errors as u1db_errors
 from u1db.backends import sqlite_backend
 
 from hashlib import sha256
-from contextlib import contextmanager
-from collections import defaultdict
 from functools import partial
 
 from pysqlcipher import dbapi2 as sqlcipher_dbapi2
@@ -427,7 +424,6 @@ class SQLCipherU1DBSync(SQLCipherDatabase):
     A dictionary that hold locks which avoid multiple sync attempts from the
     same database replica.
     """
-    syncing_lock = defaultdict(threading.Lock)
 
     def __init__(self, opts, soledad_crypto, replica_uid, cert_file,
                  defer_encryption=False, sync_db=None, sync_enc_pool=None):
@@ -532,46 +528,17 @@ class SQLCipherU1DBSync(SQLCipherDatabase):
             before the synchronisation was performed.
         :rtype: Deferred
         """
-        # the following context manager blocks until the syncing lock can be
-        # acquired.
-        with self._syncer(url, creds=creds) as syncer:
+        syncer = self._get_syncer(url, creds=creds)
 
-            def _record_received_docs(result):
-                # beware, closure. syncer is in scope.
-                self.received_docs = syncer.received_docs
-                return result
+        def _record_received_docs(result):
+            # beware, closure. syncer is in scope.
+            self.received_docs = syncer.received_docs
+            return result
 
-            # XXX could mark the critical section here...
-            d = syncer.sync(defer_decryption=defer_decryption)
-            d.addCallback(_record_received_docs)
-            return d
-
-    @contextmanager
-    def _syncer(self, url, creds=None):
-        """
-        Accesor for synchronizer.
-
-        As we reuse the same synchronizer for every sync, there can be only
-        one instance synchronizing the same database replica at the same time.
-        Because of that, this method blocks until the syncing lock can be
-        acquired.
-
-        :param creds: optional dictionary giving credentials to authorize the
-                      operation with the server.
-        :type creds: dict
-        """
-        with self.syncing_lock[self._path]:
-            syncer = self._get_syncer(url, creds=creds)
-            yield syncer
-
-    @property
-    def syncing(self):
-        lock = self.syncing_lock[self._path]
-        acquired_lock = lock.acquire(False)
-        if acquired_lock is False:
-            return True
-        lock.release()
-        return False
+        # XXX could mark the critical section here...
+        d = syncer.sync(defer_decryption=defer_decryption)
+        d.addCallback(_record_received_docs)
+        return d
 
     def _get_syncer(self, url, creds=None):
         """
diff --git a/common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py b/common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py
index 8f7af8c9..be36ddee 100644
--- a/common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py
+++ b/common/src/leap/soledad/common/tests/test_couch_operations_atomicity.py
@@ -96,6 +96,7 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
             replica_uid='replica',
             ensure_ddocs=True)
         self.tempdir = tempfile.mkdtemp(prefix="leap_tests-")
+        self.startTwistedServer()
 
     def tearDown(self):
         self.db.delete_database()
@@ -164,7 +165,6 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
         """
         Assert that the sync_log increases accordingly with sequential syncs.
         """
-        self.startServer()
         sol = self._soledad_instance(
             auth_token='auth-token',
             server_url=self.getURL())
@@ -321,8 +321,6 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
         """
         docs = []
 
-        self.startServer()
-
         sol = self._soledad_instance(
             auth_token='auth-token',
             server_url=self.getURL())
@@ -357,6 +355,7 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
 
         return d
 
+    @defer.inlineCallbacks
     def test_concurrent_syncs_do_not_fail(self):
         """
         Assert that concurrent attempts to sync end up being executed
@@ -364,8 +363,6 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
         """
         docs = []
 
-        self.startServer()
-
         sol = self._soledad_instance(
             auth_token='auth-token',
             server_url=self.getURL())
@@ -374,21 +371,15 @@ class CouchAtomicityTestCase(CouchDBTestCase, TestCaseWithServer):
         for i in xrange(0, REPEAT_TIMES):
             d = sol.create_doc({})
             d.addCallback(lambda doc: docs.append(doc.doc_id))
-            d.addCallback(lambda _: sol.sync())
+            d.addCallback(sol.sync)
             deferreds.append(d)
+        yield defer.gatherResults(deferreds, consumeErrors=True)
 
-        def _assert_logs(results):
-            transaction_log = self.db._get_transaction_log()
-            self.assertEqual(REPEAT_TIMES, len(transaction_log))
-            # assert all documents are in the remote log
-            self.assertEqual(REPEAT_TIMES, len(docs))
-            for doc_id in docs:
-                self.assertEqual(
-                    1,
-                    len(filter(lambda t: t[0] == doc_id, transaction_log)))
-
-        d = defer.gatherResults(deferreds)
-        d.addCallback(_assert_logs)
-        d.addCallback(lambda _: sol.close())
-
-        return d
+        transaction_log = self.db._get_transaction_log()
+        self.assertEqual(REPEAT_TIMES, len(transaction_log))
+        # assert all documents are in the remote log
+        self.assertEqual(REPEAT_TIMES, len(docs))
+        for doc_id in docs:
+            self.assertEqual(
+                1,
+                len(filter(lambda t: t[0] == doc_id, transaction_log)))
diff --git a/common/src/leap/soledad/common/tests/test_server.py b/common/src/leap/soledad/common/tests/test_server.py
index 848b68e9..17827486 100644
--- a/common/src/leap/soledad/common/tests/test_server.py
+++ b/common/src/leap/soledad/common/tests/test_server.py
@@ -347,7 +347,7 @@ class EncryptedSyncTestCase(
         Test the complete syncing chain between two soledad dbs using a
         Soledad server backed by a couch database.
         """
-        self.startServer()
+        self.startTwistedServer()
         user = 'user-' + uuid4().hex
 
         # instantiate soledad and create a document
diff --git a/common/src/leap/soledad/common/tests/test_soledad.py b/common/src/leap/soledad/common/tests/test_soledad.py
index 8c791672..1217b763 100644
--- a/common/src/leap/soledad/common/tests/test_soledad.py
+++ b/common/src/leap/soledad/common/tests/test_soledad.py
@@ -361,7 +361,7 @@ class SoledadSignalingTestCase(BaseSoledadTest):
 
         def _assert_done_data_sync_signal_emitted(results):
             # assert the signal has been emitted
-            soledad.client.signal.assert_called_with(
+            soledad.client.events.emit.assert_called_with(
                 catalog.SOLEDAD_DONE_DATA_SYNC,
                 ADDRESS,
             )
diff --git a/common/src/leap/soledad/common/tests/util.py b/common/src/leap/soledad/common/tests/util.py
index df405010..9800078d 100644
--- a/common/src/leap/soledad/common/tests/util.py
+++ b/common/src/leap/soledad/common/tests/util.py
@@ -275,7 +275,7 @@ class BaseSoledadTest(BaseLeapTest, MockedSharedDBTest):
             MockSharedDB = self.get_default_shared_mock(
                 _put_doc_side_effect)
 
-        return Soledad(
+        soledad = Soledad(
             user,
             passphrase,
             secrets_path=os.path.join(
@@ -287,6 +287,8 @@ class BaseSoledadTest(BaseLeapTest, MockedSharedDBTest):
             defer_encryption=self.defer_sync_encryption,
             shared_db=MockSharedDB(),
             auth_token=auth_token)
+        self.addCleanup(soledad.close)
+        return soledad
 
     def assertGetEncryptedDoc(
             self, db, doc_id, doc_rev, content, has_conflicts):