sjs/vdirsyncer
1
0
Fork 0
mirror of https://github.com/samsonjs/vdirsyncer.git synced 2026-03-26 09:05:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'click'

Browse source 
Fix #80
This commit is contained in:
Markus Unterwaditzer 2014-06-19 18:16:23 +02:00
commit 2cc375aee6
8 changed files with 225 additions and 79 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
setup.py
View file

@ -38,7 +38,7 @@ setup(
'console_scripts': ['vdirsyncer = vdirsyncer.cli:main']
},
install_requires=[
'argvard>=0.3.0',
'click',
'requests',
'lxml',
'icalendar>=3.6',

4
tests/__init__.py
View file

@ -15,6 +15,10 @@ from vdirsyncer.utils.vobject import normalize_item as _normalize_item
vdirsyncer.log.set_level(vdirsyncer.log.logging.DEBUG)
def blow_up(*a, **kw):
raise AssertionError('Did not expect to be called.')
def normalize_item(item):
if not isinstance(item, text_type):
item = item.raw

76
tests/test_cli.py
View file

@ -8,6 +8,8 @@
'''
from textwrap import dedent
from click.testing import CliRunner
import vdirsyncer.cli as cli
@ -124,3 +126,77 @@ def test_parse_pairs_args():
('one', 'c'),
('eins', None)
]
def test_simple_run(tmpdir):
config_file = tmpdir.join('config')
config_file.write(dedent('''
[general]
status_path = {0}/status/
[pair my_pair]
a = my_a
b = my_b
[storage my_a]
type = filesystem
path = {0}/path_a/
fileext = .txt
[storage my_b]
type = filesystem
path = {0}/path_b/
fileext = .txt
''').format(str(tmpdir)))
runner = CliRunner(env={'VDIRSYNCER_CONFIG': str(config_file)})
result = runner.invoke(cli.app, ['sync'])
assert not result.exception
assert result.output.lower().strip() == 'syncing my_pair'
tmpdir.join('path_a/haha.txt').write('UID:haha')
result = runner.invoke(cli.app, ['sync'])
assert tmpdir.join('path_b/haha.txt').read() == 'UID:haha'
def test_missing_general_section(tmpdir):
config_file = tmpdir.join('config')
config_file.write(dedent('''
[pair my_pair]
a = my_a
b = my_b
[storage my_a]
type = filesystem
path = {0}/path_a/
fileext = .txt
[storage my_b]
type = filesystem
path = {0}/path_b/
fileext = .txt
''').format(str(tmpdir)))
runner = CliRunner()
result = runner.invoke(
cli.app, ['sync'],
env={'VDIRSYNCER_CONFIG': str(config_file)}
)
assert result.exception
assert 'critical: unable to find general section' in result.output.lower()
def test_verbosity(tmpdir):
runner = CliRunner()
config_file = tmpdir.join('config')
config_file.write(dedent('''
[general]
status_path = {0}/status/
''').format(str(tmpdir)))
result = runner.invoke(
cli.app, ['--verbosity=HAHA', 'sync'],
env={'VDIRSYNCER_CONFIG': str(config_file)}
)
assert result.exception
assert 'invalid verbosity value'

41
tests/utils/test_main.py
View file

@ -7,11 +7,13 @@
:license: MIT, see LICENSE for more details.
'''
import click
from click.testing import CliRunner
import pytest
import vdirsyncer.utils as utils
from vdirsyncer.utils.vobject import split_collection
from .. import normalize_item, SIMPLE_TEMPLATE, BARE_EVENT_TEMPLATE
from .. import blow_up, normalize_item, SIMPLE_TEMPLATE, BARE_EVENT_TEMPLATE
def test_parse_options():
@ -58,7 +60,7 @@ def test_get_password_from_netrc(monkeypatch):
return username, 'bogus', password
monkeypatch.setattr('netrc.netrc', Netrc)
monkeypatch.setattr('getpass.getpass', None)
monkeypatch.setattr('getpass.getpass', blow_up)
_password = utils.get_password(username, resource)
assert _password == password
@ -101,13 +103,46 @@ def test_get_password_from_system_keyring(monkeypatch, resources_to_test):
return None
monkeypatch.setattr('netrc.netrc', Netrc)
monkeypatch.setattr('getpass.getpass', None)
monkeypatch.setattr('getpass.getpass', blow_up)
_password = utils.get_password(username, resource)
assert _password == password
assert netrc_calls == [hostname]
def test_get_password_from_prompt(monkeypatch):
getpass_calls = []
class Netrc(object):
def authenticators(self, hostname):
return None
class Keyring(object):
def get_password(self, *a, **kw):
return None
monkeypatch.setattr('netrc.netrc', Netrc)
monkeypatch.setattr(utils, 'keyring', Keyring())
user = 'my_user'
resource = 'http://example.com'
@click.command()
def fake_app():
x = utils.get_password(user, resource)
click.echo('Password is {}'.format(x))
runner = CliRunner()
result = runner.invoke(fake_app, input='my_password\n\n')
assert not result.exception
assert result.output.splitlines() == [
'Server password for {} at the resource {}: '.format(user, resource),
'Save this password in the keyring? [y/N]: ',
'Password is my_password'
]
def test_get_class_init_args():
class Foobar(object):
def __init__(self, foo, bar, baz=None):

112
vdirsyncer/cli.py
View file

@ -7,11 +7,12 @@
:license: MIT, see LICENSE for more details.
'''
import functools
import json
import os
import sys
import argvard
import click
from . import log
from .storage import storage_names
@ -169,14 +170,6 @@ def expand_collection(pair, collection, all_pairs, all_storages):
return [collection]
def main():
env = os.environ
fname = expand_path(env.get('VDIRSYNCER_CONFIG', '~/.vdirsyncer/config'))
cfg = load_config(fname)
_main(env, cfg)
def parse_pairs_args(pairs_args, all_pairs):
if not pairs_args:
pairs_args = list(all_pairs)
@ -204,43 +197,55 @@ def parse_pairs_args(pairs_args, all_pairs):
yield pair, c
def _main(env, file_cfg):
general, all_pairs, all_storages = file_cfg
app = argvard.Argvard()
def _create_app():
def catch_errors(f):
@functools.wraps(f)
def inner(*a, **kw):
try:
f(*a, **kw)
except CliError as e:
cli_logger.critical(str(e))
sys.exit(1)
@app.option('--verbosity verbosity')
def verbose_option(context, verbosity):
'''
Basically Python logging levels.
return inner
CRITICAL: Config errors, at most.
ERROR: Normal errors, at most.
WARNING: Problems of which vdirsyncer thinks that it can handle them
itself, but which might crash other clients.
INFO: Normal output.
DEBUG: Show e.g. HTTP traffic. Not supposed to be readable by the
normal user.
'''
verbosity = verbosity.upper()
x = getattr(log.logging, verbosity, None)
def validate_verbosity(ctx, param, value):
x = getattr(log.logging, value.upper(), None)
if x is None:
raise ValueError(u'Invalid verbosity value: {}'.format(verbosity))
log.set_level(x)
raise click.BadParameter('Invalid verbosity value {}. Must be '
'CRITICAL, ERROR, WARNING, INFO or DEBUG'
.format(value))
return x
sync_command = argvard.Command()
@click.group()
@click.option('--verbosity', '-v', default='INFO',
callback=validate_verbosity,
help='Either CRITICAL, ERROR, WARNING, INFO or DEBUG')
@click.pass_context
@catch_errors
def app(ctx, verbosity):
'''
vdirsyncer -- synchronize calendars and contacts
'''
log.add_handler(log.stdout_handler)
log.set_level(verbosity)
@sync_command.option('--force-delete status_name')
def force_delete(context, status_name):
'''Pretty please delete all my data.'''
context.setdefault('force_delete', set()).add(status_name)
if ctx.obj is None:
ctx.obj = {}
@sync_command.main('[pairs...]')
def sync_main(context, pairs=None):
if 'config' not in ctx.obj:
fname = expand_path(os.environ.get('VDIRSYNCER_CONFIG',
'~/.vdirsyncer/config'))
ctx.obj['config'] = load_config(fname)
@app.command()
@click.argument('pairs', nargs=-1)
@click.option('--force-delete', multiple=True,
help=('Disable data-loss protection for the given pairs. '
'Can be passed multiple times'))
@click.pass_context
@catch_errors
def sync(ctx, pairs, force_delete):
'''
Synchronize the given pairs. If no pairs are given, all will be
synchronized.
@ -251,9 +256,11 @@ def _main(env, file_cfg):
`vdirsyncer sync bob/first_collection` will sync "first_collection"
from the pair "bob".
'''
general, all_pairs, all_storages = ctx.obj['config']
actions = []
handled_collections = set()
force_delete = context.get('force_delete', set())
force_delete = set(force_delete)
for pair_name, _collection in parse_pairs_args(pairs, all_pairs):
for collection in expand_collection(pair_name, _collection,
all_pairs, all_storages):
@ -287,23 +294,20 @@ def _main(env, file_cfg):
if processes == 1:
cli_logger.debug('Not using multiprocessing.')
map(_sync_collection, actions)
rv = (_sync_collection(x) for x in actions)
else:
cli_logger.debug('Using multiprocessing.')
from multiprocessing import Pool
p = Pool(processes=general.get('processes', 0) or len(actions))
if not all(p.map_async(_sync_collection, actions).get(10**9)):
raise CliError()
rv = p.map_async(_sync_collection, actions).get(10**9)
app.register_command('sync', sync_command)
if not all(rv):
sys.exit(1)
try:
app()
except CliError as e:
msg = str(e)
if msg:
cli_logger.critical(msg)
sys.exit(1)
return app
app = main = _create_app()
del _create_app
def _sync_collection(x):
@ -330,7 +334,7 @@ def sync_collection(config_a, config_b, pair_name, collection, pair_options,
)
except StorageEmpty as e:
rv = False
cli_logger.critical(
cli_logger.error(
'{collection}: Storage "{side}" ({storage}) was completely '
'emptied. Use "--force-delete {status_name}" to synchronize that '
'emptyness to the other side, or delete the status by yourself to '
@ -343,7 +347,7 @@ def sync_collection(config_a, config_b, pair_name, collection, pair_options,
)
except SyncConflict as e:
rv = False
cli_logger.critical(
cli_logger.error(
'{collection}: One item changed on both sides. Resolve this '
'conflict manually, or by setting the `conflict_resolution` '
'parameter in your config file.\n'

37
vdirsyncer/log.py
View file

@ -7,17 +7,48 @@
:license: MIT, see LICENSE for more details.
'''
import logging
import sys
import click
stdout_handler = logging.StreamHandler(sys.stdout)
class ColorFormatter(logging.Formatter):
colors = {
'error': dict(fg='red'),
'exception': dict(fg='red'),
'critical': dict(fg='red'),
'debug': dict(fg='blue'),
'warning': dict(fg='yellow')
}
def format(self, record):
if not record.exc_info:
level = record.levelname.lower()
if level in self.colors:
prefix = click.style('{}: '.format(level),
**self.colors[level])
record.msg = prefix + record.msg
return logging.Formatter.format(self, record)
class ClickStream(object):
def write(self, string):
click.echo(string, nl=False)
stdout_handler = logging.StreamHandler(ClickStream())
stdout_handler.formatter = ColorFormatter()
default_level = logging.INFO
def add_handler(handler):
for logger in loggers.values():
logger.addHandler(handler)
def create_logger(name):
x = logging.getLogger(name)
x.setLevel(default_level)
x.addHandler(stdout_handler)
return x

26
vdirsyncer/utils/__init__.py
View file

@ -9,10 +9,12 @@
import os
import click
import requests
from .. import exceptions, log
from .compat import get_raw_input, iteritems, urlparse
from .compat import iteritems, urlparse
logger = log.get(__name__)
@ -116,7 +118,9 @@ def _password_from_keyring(username, resource):
parsed = urlparse.urlsplit(key)
path = parsed.path
if path.endswith('/'):
if not path:
return None
elif path.endswith('/'):
path = path.rstrip('/')
else:
path = path.rsplit('/', 1)[0] + '/'
@ -155,8 +159,6 @@ def get_password(username, resource):
"""
import getpass
for func in (_password_from_netrc, _password_from_keyring):
password = func(username, resource)
if password is not None:
@ -164,18 +166,14 @@ def get_password(username, resource):
.format(username, func.__doc__))
return password
prompt = ('Server password for {} at the resource {}: '
prompt = ('Server password for {} at the resource {}'
.format(username, resource))
password = getpass.getpass(prompt=prompt)
password = click.prompt(prompt, hide_input=True)
if keyring is not None:
answer = None
while answer not in ['', 'y', 'n']:
prompt = 'Save this password in the keyring? [y/N] '
answer = get_raw_input(prompt).lower()
if answer == 'y':
keyring.set_password(password_key_prefix + resource,
username, password)
if keyring is not None and \
click.confirm('Save this password in the keyring?', default=False):
keyring.set_password(password_key_prefix + resource,
username, password)
return password

6
vdirsyncer/utils/compat.py
View file

@ -12,7 +12,7 @@ import sys
PY2 = sys.version_info[0] == 2
if PY2:
if PY2: # pragma: no cover
import urlparse
from urllib import \
quote_plus as urlquote_plus, \
@ -20,12 +20,10 @@ if PY2:
text_type = unicode # flake8: noqa
iteritems = lambda x: x.iteritems()
itervalues = lambda x: x.itervalues()
get_raw_input = raw_input
else:
else: # pragma: no cover
import urllib.parse as urlparse
urlquote_plus = urlparse.quote_plus
urlunquote_plus = urlparse.unquote_plus
text_type = str
iteritems = lambda x: x.items()
itervalues = lambda x: x.values()
get_raw_input = input