Skip to content

Instantly share code, notes, and snippets.

@indradhanush
Last active August 29, 2015 14:05
Show Gist options
  • Save indradhanush/5c3966167c9adf069af7 to your computer and use it in GitHub Desktop.
Save indradhanush/5c3966167c9adf069af7 to your computer and use it in GitHub Desktop.
Using mock
class ZMQSynchronizer(Synchronizer):
"""
Extends leap.soledad.client.sync.SoledadSynchronizer
"""
syncing_lock = Lock()
def _sync(self, autocreate=False, defer_decryption=True):
"""
Overrides SoledadSynchronizer._sync
"""
self.source_replica_uid = self.source._replica_uid
(target_latest_generation, target_latest_trans_id,
ensure_callback) = self._sync_step_get_sync_info(autocreate)
self._sync_step_validate_info_at_target()
docs_by_generation = self._sync_step_find_changes_to_send(
target_latest_generation, target_latest_trans_id)
source_current_gen = self._sync_step_sync_exchange(
docs_by_generation, ensure_callback)
return source_current_gen
def _sync_step_get_sync_info(self, autocreate):
"""
Method to go through the get_sync_info step of the entire sync
process.
"""
# get_sync_info
# source_last_known_generation is the last generation of the
# source that the target knows of. Similarly for
# source_last_known_trans_id
ensure_callback = None
sync_target = self.sync_target
try:
(self.target_replica_uid, target_latest_generation,
target_latest_trans_id,
sync_target.source_last_known_generation,
sync_target.source_last_known_trans_id) =\
sync_target.get_sync_info(self.source_replica_uid)
except DatabaseDoesNotExist:
if not autocreate:
raise
else:
# Never synced yet.
self.target_replica_uid = None
(target_latest_generation,
target_latest_generation) = (0, "")
(sync_target.source_last_known_generation,
sync_target.source_last_known_trans_id) = (0, "")
if self.target_replica_uid is None:
def ensure_callback(replica_uid):
self.target_replica_uid = replica_uid
if self.target_replica_uid == self.source_replica_uid:
raise InvalidReplicaUID
return target_latest_generation, target_latest_trans_id, ensure_callback
def _sync_step_validate_info_at_target(self):
# Validate last known info about source present at target.
sync_target = self.sync_target
self.source.validate_gen_and_trans_id(
sync_target.source_last_known_generation,
sync_target.source_last_known_trans_id)
def _sync_step_find_changes_to_send(self, target_latest_generation,
target_latest_trans_id):
sync_target = self.sync_target
# Find changes at source to send to target.
(sync_target.source_current_gen,
sync_target.source_current_trans_id,
changes) = self.source.whats_changed(
sync_target.source_last_known_generation)
if self.target_replica_uid is None:
(sync_target.target_last_known_generation,
sync_target.target_last_known_trans_id) = (0, "")
else:
(sync_target.target_last_known_generation,
sync_target.target_last_known_trans_id) =\
self.source._get_replica_gen_and_trans_id(
self.target_replica_uid)
if not changes and\
sync_target.target_last_known_generation == target_latest_generation:
if sync_target.target_last_known_trans_id != target_latest_trans_id:
raise InvalidTransactionId
return sync_target.source_current_gen
changed_doc_ids = [doc_id for doc_id, _, _ in changes]
docs_to_send = self.source.get_docs(changed_doc_ids,
check_for_conflicts=False,
include_deleted=True)
docs_by_generation = []
i = 0
for doc in docs_to_send:
_, gen, trans_id = changes[i]
docs_by_generation.append((doc, gen, trans_id))
i += 1
return docs_by_generation
def _sync_step_sync_exchange(self, docs_by_generation, ensure_callback):
# sync_exchange
sync_target = self.sync_target
try:
# TODO: Returns the new target gen and trans_id but it is
# already set as sync_target.target_last_known_generation
# and sync_target.target_last_known_trans_id ; Discuss?
new_gen, new_trans_id = sync_target.sync_exchange(
docs_by_generation, self.source_replica_uid,
self._insert_doc_from_target, ensure_callback=ensure_callback)
# TODO: Override complete_sync to change the param being
# passed from "my_gen" to "source_current_gen" ; Trivial.
# May not be required.
self._syncing_info = {
TARGET_REPLICA_UID_KEY: self.target_replica_uid,
"new_gen": new_gen,
"new_trans_id": new_trans_id,
"my_gen": sync_target.source_current_gen
}
# TODO: Implement defer_decryption in ZMQSyncTarget
# if defer_decryption and not sync_target.has_syncdb():
# defer_decryption = False
self.complete_sync()
except Exception as e:
# TODO: Log this.
print e
finally:
sync_target.stop()
return sync_target.source_current_gen
class ZMQSynchronizerTest(BaseSynchronizerTest):
"""
Test suite for zmq_transport.client.sync.ZMQSynchronizer
"""
def setUp(self):
self.sync_target = ZMQSyncTarget([ENDPOINT_CLIENT_HANDLER,
ENDPOINT_PUBLISHER])
self.source = u1db.open(os.path.join(
DATABASE_ROOT, "source-USER-1.u1db"), create=False)
# self.patcher = patch("zmq_transport.client.sync.ZMQSynchronizer", spec=True)
# self.patcher.start()
def test__sync(self):
"""
Tests ZMQSynchronizer._sync
"""
synchronizer = ZMQSynchronizer(self.source, self.sync_target)
synchronizer._sync_step_get_sync_info = MagicMock(
name="_sync_step_get_sync_info")
synchronizer._sync_step_get_sync_info.return_value = (0, "", None)
synchronizer._sync_step_validate_info_at_target = MagicMock(
name="_sync_step_validate_info_at_target")
synchronizer._sync_step_find_changes_to_send = MagicMock(
name="_sync_step_find_changes_to_send")
synchronizer._sync_step_find_changes_to_send.return_value = []
synchronizer._sync_step_sync_exchange = MagicMock(
name="_sync_step_sync_exchange")
synchronizer._sync_step_sync_exchange.return_value = 0
synchronizer._sync()
synchronizer._sync_step_get_sync_info.assert_called_once_with(False)
synchronizer._sync_step_validate_info_at_target.assert_called_once()
synchronizer._sync_step_find_changes_to_send.assert_called_once_with(0,
"")
synchronizer._sync_step_sync_exchange.assert_called_once_with([], None)
def tearDown(self):
self.sync_target = None
self.source.close()
# self.patcher.stop()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment