#!/usr/bin/env python3 # Copyright 2020 Collabora Ltd. # # SPDX-License-Identifier: MIT import logging import os import re import signal import subprocess import sys try: import typing typing # placate pyflakes except ImportError: pass from testutils import ( BaseTest, run_subprocess, test_main, ) logger = logging.getLogger('test-adverb') EX_USAGE = 64 class TestAdverb(BaseTest): def run_subprocess( self, args, # type: typing.Union[typing.List[str], str] check=False, input=None, # type: typing.Optional[bytes] timeout=None, # type: typing.Optional[int] **kwargs # type: typing.Any ): logger.info('Running: %r', args) return run_subprocess( args, check=check, input=input, timeout=timeout, **kwargs ) def setUp(self) -> None: super().setUp() if 'PRESSURE_VESSEL_UNINSTALLED' in os.environ: self.adverb = self.command_prefix + [ 'env', '-u', 'LD_AUDIT', '-u', 'LD_PRELOAD', # In unit tests it isn't straightforward to find the real # ${PLATFORM}, so we use a predictable mock implementation # that always uses PvMultiarchDetails.platforms[0]. 'PRESSURE_VESSEL_TEST_STANDARDIZE_PLATFORM=1', os.path.join( self.top_builddir, 'pressure-vessel', 'pressure-vessel-adverb' ), ] self.helper = self.command_prefix + [ os.path.join( self.top_builddir, 'tests', 'pressure-vessel', 'test-helper' ), ] else: self.skipTest('Not available as an installed-test') def test_ld_preload(self) -> None: completed = run_subprocess( self.adverb + [ '--ld-audit=/nonexistent/libaudit.so', '--ld-preload=/nonexistent/libpreload.so', '--ld-preload=/nonexistent/ubuntu12_32/gameoverlayrenderer.so', '--ld-preload=/nonexistent/ubuntu12_64/gameoverlayrenderer.so', '--', 'sh', '-euc', # The hard-coded i686 and xeon_phi here must match up with # pv_multiarch_details[i].platforms[0], which is what is used # as a mock implementation under # PRESSURE_VESSEL_TEST_STANDARDIZE_PLATFORM=1. r''' ld_audit="$LD_AUDIT" ld_preload="$LD_PRELOAD" unset LD_AUDIT unset LD_PRELOAD echo "LD_AUDIT=$ld_audit" echo "LD_PRELOAD=$ld_preload" IFS=: for item in $ld_preload; do case "$item" in (*\$\{PLATFORM\}*) i686="$(echo "$item" | sed -e 's/[$]{PLATFORM}/i686/g')" xeon_phi="$(echo "$item" | sed -e 's/[$]{PLATFORM}/xeon_phi/g')" printf "i686: symlink to " readlink "$i686" printf "xeon_phi: symlink to " readlink "$xeon_phi" ;; (*) echo "literal $item" ;; esac done ''', ], check=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=2, universal_newlines=True, ) stdout = completed.stdout assert stdout is not None lines = stdout.splitlines() self.assertEqual( lines[0], 'LD_AUDIT=/nonexistent/libaudit.so', ) self.assertEqual( re.sub(r':/[^:]*?/pressure-vessel-libs-....../', r':/tmp/pressure-vessel-libs-XXXXXX/', lines[1]), ('LD_PRELOAD=/nonexistent/libpreload.so' ':/tmp/pressure-vessel-libs-XXXXXX/' '${PLATFORM}/gameoverlayrenderer.so') ) self.assertEqual(lines[2], 'literal /nonexistent/libpreload.so') self.assertEqual( lines[3], 'i686: symlink to /nonexistent/ubuntu12_32/gameoverlayrenderer.so', ) self.assertEqual( lines[4], ('xeon_phi: symlink to ' '/nonexistent/ubuntu12_64/gameoverlayrenderer.so'), ) def test_stdio_passthrough(self) -> None: proc = subprocess.Popen( self.adverb + [ '--', 'sh', '-euc', ''' if [ "$(cat)" != "hello, world!" ]; then exit 1 fi echo $$ exec >/dev/null exec sleep infinity ''', ], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=2, universal_newlines=True, ) pid = 0 try: stdin = proc.stdin assert stdin is not None stdin.write('hello, world!') stdin.close() stdout = proc.stdout assert stdout is not None pid = int(stdout.read().strip()) finally: if pid: os.kill(pid, signal.SIGTERM) else: proc.terminate() self.assertIn( proc.wait(), (128 + signal.SIGTERM, -signal.SIGTERM), ) def test_fd_passthrough(self) -> None: read_end, write_end = os.pipe2(os.O_CLOEXEC) proc = subprocess.Popen( self.adverb + [ '--pass-fd=%d' % write_end, '--', 'sh', '-euc', 'echo hello >&%d' % write_end, ], pass_fds=[write_end], stdout=2, stderr=2, universal_newlines=True, ) try: os.close(write_end) with os.fdopen(read_end, 'rb') as reader: self.assertEqual(reader.read(), b'hello\n') finally: proc.wait() self.assertEqual(proc.returncode, 0) def test_wrong_options(self) -> None: for option in ( '--an-unknown-option', '--pass-fd=-1', '--shell=wrong', '--terminal=wrong', ): proc = subprocess.Popen( self.adverb + [ option, '--', 'sh', '-euc', 'exit 42', ], stdout=2, stderr=2, universal_newlines=True, ) proc.wait() self.assertEqual(proc.returncode, EX_USAGE) def tearDown(self) -> None: super().tearDown() if __name__ == '__main__': assert sys.version_info >= (3, 4), \ 'Python 3.4+ is required' test_main() # vi: set sw=4 sts=4 et: