mirror of
https://github.com/samsonjs/vdirsyncer.git
synced 2026-04-27 14:57:41 +00:00
Renamed evalcmd -> command, use py.test's tmpdir, allow arguments in passwordeval option.
This commit is contained in:
parent
4e895b8635
commit
64e9ef7dc3
2 changed files with 24 additions and 26 deletions
|
|
@ -10,7 +10,6 @@
|
||||||
import click
|
import click
|
||||||
|
|
||||||
from click.testing import CliRunner
|
from click.testing import CliRunner
|
||||||
from tempfile import mkstemp
|
|
||||||
import os
|
import os
|
||||||
import stat
|
import stat
|
||||||
import pytest
|
import pytest
|
||||||
|
|
@ -111,33 +110,33 @@ def test_get_password_from_system_keyring(monkeypatch):
|
||||||
assert _password == password
|
assert _password == password
|
||||||
|
|
||||||
|
|
||||||
def test_get_password_from_evalcmd():
|
def test_get_password_from_command(tmpdir):
|
||||||
username = 'my_username'
|
username = 'my_username'
|
||||||
resource = 'http://example.com'
|
resource = 'http://example.com'
|
||||||
password = 'testpassword'
|
password = 'testpassword'
|
||||||
|
filename = 'command.sh'
|
||||||
|
|
||||||
fd, temp_path = mkstemp()
|
filepath = str(tmpdir) + '/' + filename
|
||||||
os.close(fd)
|
f = open(filepath, 'w')
|
||||||
fp = open(temp_path, 'w')
|
f.write('#!/bin/sh\n'
|
||||||
fp.write('#!/bin/sh\n'
|
|
||||||
'[ "$1" != "my_username" ] && exit 1\n'
|
'[ "$1" != "my_username" ] && exit 1\n'
|
||||||
'[ "$2" != "example.com" ] && exit 1\n'
|
'[ "$2" != "example.com" ] && exit 1\n'
|
||||||
'echo "{}"'.format(password))
|
'echo "{}"'.format(password))
|
||||||
fp.close()
|
f.close()
|
||||||
st = os.stat(temp_path)
|
|
||||||
os.chmod(temp_path, st.st_mode | stat.S_IEXEC)
|
st = os.stat(filepath)
|
||||||
|
os.chmod(filepath, st.st_mode | stat.S_IEXEC)
|
||||||
|
|
||||||
@doubleclick.click.command()
|
@doubleclick.click.command()
|
||||||
@doubleclick.click.pass_context
|
@doubleclick.click.pass_context
|
||||||
def fake_app(ctx):
|
def fake_app(ctx):
|
||||||
ctx.obj = {'config' : ({'passwordeval' : temp_path},{},{})}
|
ctx.obj = {'config' : ({'passwordeval' : filepath},{},{})}
|
||||||
_password = utils.get_password(username, resource)
|
_password = utils.get_password(username, resource)
|
||||||
assert _password == password
|
assert _password == password
|
||||||
|
|
||||||
runner = CliRunner()
|
runner = CliRunner()
|
||||||
result = runner.invoke(fake_app)
|
result = runner.invoke(fake_app)
|
||||||
assert not result.exception
|
assert not result.exception
|
||||||
os.remove(temp_path)
|
|
||||||
|
|
||||||
|
|
||||||
def test_get_password_from_prompt():
|
def test_get_password_from_prompt():
|
||||||
|
|
|
||||||
|
|
@ -119,8 +119,8 @@ def get_password(username, resource, _lock=threading.Lock()):
|
||||||
|
|
||||||
with _lock:
|
with _lock:
|
||||||
host = urlparse.urlsplit(resource).hostname
|
host = urlparse.urlsplit(resource).hostname
|
||||||
for func in (_password_from_cache, _password_from_netrc,
|
for func in (_password_from_command, _password_from_cache,
|
||||||
_password_from_keyring, _password_from_evalcmd):
|
_password_from_netrc, _password_from_keyring):
|
||||||
password = func(username, host)
|
password = func(username, host)
|
||||||
if password is not None:
|
if password is not None:
|
||||||
logger.debug('Got password for {} from {}'
|
logger.debug('Got password for {} from {}'
|
||||||
|
|
@ -168,28 +168,27 @@ def _password_from_keyring(username, host):
|
||||||
return keyring.get_password(password_key_prefix + host, username)
|
return keyring.get_password(password_key_prefix + host, username)
|
||||||
|
|
||||||
|
|
||||||
def _password_from_evalcmd(username, host):
|
def _password_from_command(username, host):
|
||||||
'''evalcmd'''
|
'''command'''
|
||||||
import subprocess
|
import subprocess
|
||||||
|
|
||||||
try:
|
try:
|
||||||
general, _, _ = ctx.obj['config']
|
general, _, _ = ctx.obj['config']
|
||||||
_evalcmd = general.get('passwordeval', '')
|
_command = general['passwordeval'].split()
|
||||||
except (KeyError, IndexError):
|
except (IndexError, KeyError):
|
||||||
return None
|
return None
|
||||||
|
|
||||||
if _evalcmd == '':
|
command = [expand_path(_command[0])]
|
||||||
return None
|
if len(_command) > 1:
|
||||||
|
command += _command[1:]
|
||||||
evalcmd = expand_path(_evalcmd)
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
proc = subprocess.Popen([evalcmd, username, host],
|
proc = subprocess.Popen(command + [username, host],
|
||||||
stdout=subprocess.PIPE)
|
stdout=subprocess.PIPE)
|
||||||
password = proc.stdout.read().strip()
|
password = proc.stdout.read().decode('utf-8').strip()
|
||||||
except OSError, e:
|
except OSError as e:
|
||||||
logger.debug('Failed to execute evalcmd: {}\n{}'.
|
logger.debug('Failed to execute command: {}\n{}'.
|
||||||
format(evalcmd, str(e)))
|
format(" ".join(command), str(e)))
|
||||||
return None
|
return None
|
||||||
|
|
||||||
return password
|
return password
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue