How to use vdirsyncer - 10 common examples

~~~~~~~~~~~~~~~~~~~~~~~~~~~

    Using utilities from paste to wrap the PHP application into WSGI.

    :copyright: (c) 2014 Markus Unterwaditzer
    :license: MIT, see LICENSE for more details.
'''

from vdirsyncer.utils import expand_path
import subprocess
import os
import time
import pytest
import requests

owncloud_repo = expand_path(os.path.join(
    os.path.dirname(__file__), '../../../owncloud-testserver/'
))

php_sh = os.path.abspath(os.path.join(owncloud_repo, 'php.sh'))


def wait():
    for i in range(10):
        try:
            requests.get('http://127.0.0.1:8080/')
        except requests.exceptions.ConnectionError:
            time.sleep(1)
        else:
            return True
    return False

def test_conflict_resolution_invalid_mode():
    a = MemoryStorage()
    b = MemoryStorage()
    item_a = Item('UID:1\nitem a')
    item_b = Item('UID:1\nitem b')
    a.upload(item_a)
    b.upload(item_b)
    with pytest.raises(ValueError):
        sync(a, b, {}, conflict_resolution='yolo')

def test_insert_hash():
    a = MemoryStorage()
    b = MemoryStorage()
    status = {}

    item = Item('UID:1')
    href, etag = a.upload(item)
    sync(a, b, status)

    for d in status['1']:
        del d['hash']

    a.update(href, Item('UID:1\nHAHA:YES'), etag)
    sync(a, b, status)
    assert 'hash' in status['1'][0] and 'hash' in status['1'][1]

def test_conflict_resolution_both_etags_new(winning_storage):
    a = MemoryStorage()
    b = MemoryStorage()
    item = Item('UID:1')
    href_a, etag_a = a.upload(item)
    href_b, etag_b = b.upload(item)
    status = {}
    sync(a, b, status)
    assert status
    item_a = Item('UID:1\nitem a')
    item_b = Item('UID:1\nitem b')
    a.update(href_a, item_a, etag_a)
    b.update(href_b, item_b, etag_b)
    with pytest.raises(SyncConflict):
        sync(a, b, status)
    sync(a, b, status, conflict_resolution='{} wins'.format(winning_storage))
    assert items(a) == items(b) == {
        item_a.raw if winning_storage == 'a' else item_b.raw
    }

def test_missing_status_and_different_items():
    a = MemoryStorage()
    b = MemoryStorage()

    status = {}
    item1 = Item('UID:1\nhaha')
    item2 = Item('UID:1\nhoho')
    a.upload(item1)
    b.upload(item2)
    with pytest.raises(SyncConflict):
        sync(a, b, status)
    assert not status
    sync(a, b, status, conflict_resolution='a wins')
    assert items(a) == items(b) == {item1.raw}

def test_already_synced():
    a = MemoryStorage(fileext='.a')
    b = MemoryStorage(fileext='.b')
    item = Item('UID:1')
    a.upload(item)
    b.upload(item)
    status = {
        '1': ({
            'href': '1.a',
            'hash': item.hash,
            'etag': a.get('1.a')[1]
        }, {
            'href': '1.b',
            'hash': item.hash,
            'etag': b.get('1.b')[1]
        })
    }
    old_status = deepcopy(status)
    a.update = b.update = a.upload = b.upload = \
        lambda *a, **kw: pytest.fail('Method shouldn\'t have been called.')

def test_read_only_and_prefetch():
    a = MemoryStorage()
    b = MemoryStorage()
    b.read_only = True

    status = {}
    item1 = Item('UID:1\nhaha')
    item2 = Item('UID:2\nhoho')
    a.upload(item1)
    a.upload(item2)

    sync(a, b, status, force_delete=True)
    sync(a, b, status, force_delete=True)

    assert not items(a) and not items(b)

def test_upload_and_update():
    a = MemoryStorage(fileext='.a')
    b = MemoryStorage(fileext='.b')
    status = {}

    item = Item('UID:1')  # new item 1 in a
    a.upload(item)
    sync(a, b, status)
    assert items(b) == items(a) == {item.raw}

    item = Item('UID:1\nASDF:YES')  # update of item 1 in b
    b.update('1.b', item, b.get('1.b')[1])
    sync(a, b, status)
    assert items(b) == items(a) == {item.raw}

    item2 = Item('UID:2')  # new item 2 in b
    b.upload(item2)
    sync(a, b, status)
    assert items(b) == items(a) == {item.raw, item2.raw}

    item2 = Item('UID:2\nASDF:YES')  # update of item 2 in a
    a.update('2.a', item2, a.get('2.a')[1])
    sync(a, b, status)
    assert items(b) == items(a) == {item.raw, item2.raw}

def test_unicode_hrefs():
    a = MemoryStorage()
    b = MemoryStorage()
    status = {}
    href, etag = a.upload(Item('UID:äää'))
    sync(a, b, status)

def test_too_long_uid(self, tmpdir):
        s = self.storage_class(str(tmpdir), '.txt')
        item = Item('UID:' + 'hue' * 600)
        href, etag = s.upload(item)
        assert item.uid not in href