Method sync

tests/unit/sync/test_sync.py:583–632

(self, status, a, b, force_delete, conflict_resolution,
             with_error_callback, partial_sync)

        ))
    )
    def sync(self, status, a, b, force_delete, conflict_resolution,
             with_error_callback, partial_sync):
        assume(a is not b)
        old_items_a = items(a)
        old_items_b = items(b)

        a.instance_name = 'a'
        b.instance_name = 'b'

        errors = []

        if with_error_callback:
            error_callback = errors.append
        else:
            error_callback = None

        try:
            # If one storage is read-only, double-sync because changes don't
            # get reverted immediately.
            for _ in range(2 if a.read_only or b.read_only else 1):
                sync(a, b, status,
                     force_delete=force_delete,
                     conflict_resolution=conflict_resolution,
                     error_callback=error_callback,
                     partial_sync=partial_sync)

            for e in errors:
                raise e
        except PartialSync:
            assert partial_sync == 'error'
        except ActionIntentionallyFailed:
            pass
        except BothReadOnly:
            assert a.read_only and b.read_only
            assume(False)
        except StorageEmpty:
            if force_delete:
                raise
            else:
                assert not list(a.list()) or not list(b.list())
        else:
            items_a = items(a)
            items_b = items(b)

            assert items_a == items_b or partial_sync == 'ignore'
            assert items_a == old_items_a or not a.read_only
            assert items_b == old_items_b or not b.read_only

            assert set(a.items) | set(b.items) == set(status) or \
                partial_sync == 'ignore'


TestSyncMachine = SyncMachine.TestCase