gaschz/ansible-qubes
1
0
Fork 0
mirror of https://github.com/Rudd-O/ansible-qubes.git synced 2025-11-04 13:38:55 +01:00
Code Issues Packages Projects Releases Wiki Activity

Fix connection plugin tests.

Browse Source
This commit is contained in:
Rudd-O 2020-05-05 10:53:09 +00:00
parent c2f006868c
commit eb8d0ab162
2 changed files with 81 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
connection_plugins/qubes.py
View File

@ -450,7 +450,7 @@ class Connection(ConnectionBase):
super(Connection, self).fetch_file(in_path, out_path)
display.vvvv("FETCH %s to %s" % (in_path, out_path), host=self._play_context.remote_addr)
in_path = _prefix_login_path(in_path)
out_file = open(out_path, "wb")
with open(out_path, "wb") as out_file:
try:
payload = 'fetch(%r, %r)\n' % (in_path, BUFSIZE)
self._transport.stdin.write(payload.encode("utf-8"))

66
connection_plugins/test_qubes.py
View File

@ -1,81 +1,103 @@
import sys, os ; sys.path.append(os.path.dirname(__file__))
import StringIO
import qubes
import contextlib
try:
from StringIO import StringIO
BytesIO = StringIO
except ImportError:
from io import StringIO, BytesIO
import unittest
import tempfile
import qubes
cases = [
(['true'], '', 'Y\n0\n0\n0\n'),
(['false'], '', 'Y\n1\n0\n0\n'),
(['sh', '-c', 'echo yes'], '', 'Y\n0\n4\nyes\n0\n'),
(['sh', '-c', 'echo yes >&2'], '', 'Y\n0\n0\n4\nyes\n'),
]
cases_with_harness = [
if sys.version_info.major == 3:
cases = [
(['true'], '', b'Y\n0\n0\n0\n'),
(['false'], '', b'Y\n1\n0\n0\n'),
(['sh', '-c', 'echo yes'], '', b'Y\n0\n4\nyes\n0\n'),
(['sh', '-c', 'echo yes >&2'], '', b'Y\n0\n0\n4\nyes\n'),
]
cases_with_harness = [
(['true'], '', 0, '', ''),
(['false'], '', 1, '', ''),
(['sh', '-c', 'echo yes'], '', 0, 'yes\n', ''),
(['sh', '-c', 'echo yes >&2'], '', 0, '', 'yes\n'),
]
(['sh', '-c', 'echo yes'], '', 0, b'yes\n', ''),
(['sh', '-c', 'echo yes >&2'], '', 0, '', b'yes\n'),
]
else:
cases = []
cases_with_harness = []
class MockPlayContext(object):
shell = 'sh'
executable = 'sh'
become = False
become_method = 'sudo'
remote_addr = '127.0.0.7'
@contextlib.contextmanager
def local_connection():
c = qubes.Connection(
MockPlayContext(), None,
transport_cmd=['sh', '-c', '"$@"']
)
c._options = {"management_proxy": None}
return c
try:
yield c
finally:
c.close()
class TestBasicThings(unittest.TestCase):
def test_popen(self):
for cmd, in_, out in cases:
outf = StringIO.StringIO()
outf = BytesIO()
qubes.popen(cmd, in_, outf=outf)
self.assertMultiLineEqual(
self.assertEqual(
outf.getvalue(),
out
)
def test_exec_command_with_harness(self):
for cmd, in_, ret, out, err in cases_with_harness:
c = local_connection()
with local_connection() as c:
retcode, stdout, stderr = c.exec_command(cmd)
self.assertEqual(ret, retcode)
self.assertMultiLineEqual(out, stdout)
self.assertMultiLineEqual(err, stderr)
c.close()
self.assertEqual(out, stdout)
self.assertEqual(err, stderr)
self.assertEqual(c._transport, None)
def test_fetch_file_with_harness(self):
if sys.version_info.major == 2:
in_text = "abcd"
else:
in_text = b"abcd"
with tempfile.NamedTemporaryFile() as x:
x.write(in_text)
x.flush()
with tempfile.NamedTemporaryFile() as y:
c = local_connection()
with local_connection() as c:
c.fetch_file(in_path=x.name, out_path=y.name)
y.seek(0)
out_text = y.read()
self.assertEqual(in_text, out_text)
def test_put_file_with_harness(self):
if sys.version_info.major == 2:
in_text = "abcd"
else:
in_text = b"abcd"
with tempfile.NamedTemporaryFile() as x:
x.write(in_text)
x.flush()
with tempfile.NamedTemporaryFile() as y:
c = local_connection()
with local_connection() as c:
c.put_file(in_path=x.name, out_path=y.name)
y.seek(0)
out_text = y.read()
self.assertEqual(in_text, out_text)