Use Popen.communicate to get stdout from subprocess
This commit is contained in:
@@ -17,8 +17,11 @@ log = logging.getLogger(__name__)
|
|||||||
def check_output(args, env=None, sp=subprocess):
|
def check_output(args, env=None, sp=subprocess):
|
||||||
"""Call an external binary and return its stdout."""
|
"""Call an external binary and return its stdout."""
|
||||||
log.debug('calling %s with env %s', args, env)
|
log.debug('calling %s with env %s', args, env)
|
||||||
output = sp.check_output(args=args, env=env)
|
p = sp.Popen(args=args, env=env, stdin=sp.PIPE, stdout=sp.PIPE, stderr=sp.PIPE)
|
||||||
|
(output, error) = p.communicate()
|
||||||
log.debug('output: %r', output)
|
log.debug('output: %r', output)
|
||||||
|
if error:
|
||||||
|
log.debug('error: %r', error)
|
||||||
return output
|
return output
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -53,6 +53,14 @@ class FakeSocket:
|
|||||||
self.tx.write(data)
|
self.tx.write(data)
|
||||||
|
|
||||||
|
|
||||||
|
def mock_subprocess(output, error=b''):
|
||||||
|
sp = mock.Mock(spec=['Popen', 'PIPE'])
|
||||||
|
p = mock.Mock(spec=['communicate'])
|
||||||
|
sp.Popen.return_value = p
|
||||||
|
p.communicate.return_value = (output, error)
|
||||||
|
return sp
|
||||||
|
|
||||||
|
|
||||||
def test_sign_digest():
|
def test_sign_digest():
|
||||||
sock = FakeSocket()
|
sock = FakeSocket()
|
||||||
sock.rx.write(b'OK Pleased to meet you, process XYZ\n')
|
sock.rx.write(b'OK Pleased to meet you, process XYZ\n')
|
||||||
@@ -61,10 +69,8 @@ def test_sign_digest():
|
|||||||
sock.rx.seek(0)
|
sock.rx.seek(0)
|
||||||
keygrip = '1234'
|
keygrip = '1234'
|
||||||
digest = b'A' * 32
|
digest = b'A' * 32
|
||||||
sp = mock.Mock(spec=['check_output'])
|
|
||||||
sp.check_output.return_value = '/dev/pts/0'
|
|
||||||
sig = keyring.sign_digest(sock=sock, keygrip=keygrip,
|
sig = keyring.sign_digest(sock=sock, keygrip=keygrip,
|
||||||
digest=digest, sp=sp,
|
digest=digest, sp=mock_subprocess('/dev/pts/0'),
|
||||||
environ={'DISPLAY': ':0'})
|
environ={'DISPLAY': ':0'})
|
||||||
assert sig == (0x30313233343536373839414243444546,)
|
assert sig == (0x30313233343536373839414243444546,)
|
||||||
assert sock.tx.getvalue() == b'''RESET
|
assert sock.tx.getvalue() == b'''RESET
|
||||||
@@ -85,8 +91,7 @@ def test_iterlines():
|
|||||||
|
|
||||||
|
|
||||||
def test_get_agent_sock_path():
|
def test_get_agent_sock_path():
|
||||||
sp = mock.Mock(spec=['check_output'])
|
sp = mock_subprocess(b'''sysconfdir:/usr/local/etc/gnupg
|
||||||
sp.check_output.return_value = b'''sysconfdir:/usr/local/etc/gnupg
|
|
||||||
bindir:/usr/local/bin
|
bindir:/usr/local/bin
|
||||||
libexecdir:/usr/local/libexec
|
libexecdir:/usr/local/libexec
|
||||||
libdir:/usr/local/lib/gnupg
|
libdir:/usr/local/lib/gnupg
|
||||||
@@ -96,6 +101,6 @@ dirmngr-socket:/run/user/1000/gnupg/S.dirmngr
|
|||||||
agent-ssh-socket:/run/user/1000/gnupg/S.gpg-agent.ssh
|
agent-ssh-socket:/run/user/1000/gnupg/S.gpg-agent.ssh
|
||||||
agent-socket:/run/user/1000/gnupg/S.gpg-agent
|
agent-socket:/run/user/1000/gnupg/S.gpg-agent
|
||||||
homedir:/home/roman/.gnupg
|
homedir:/home/roman/.gnupg
|
||||||
'''
|
''')
|
||||||
expected = b'/run/user/1000/gnupg/S.gpg-agent'
|
expected = b'/run/user/1000/gnupg/S.gpg-agent'
|
||||||
assert keyring.get_agent_sock_path(sp=sp) == expected
|
assert keyring.get_agent_sock_path(sp=sp) == expected
|
||||||
|
|||||||
Reference in New Issue
Block a user