-
Simon McVittie authored
This avoids them colliding with steam-runtime-tools' tests. Signed-off-by:
Simon McVittie <smcv@collabora.com>
Simon McVittie authoredThis avoids them colliding with steam-runtime-tools' tests. Signed-off-by:
Simon McVittie <smcv@collabora.com>
launcher.py 20.18 KiB
#!/usr/bin/env python3
# Copyright 2020 Collabora Ltd.
#
# SPDX-License-Identifier: MIT
import logging
import os
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import uuid
try:
import typing
typing # placate pyflakes
except ImportError:
pass
from testutils import (
BaseTest,
run_subprocess,
test_main,
)
logger = logging.getLogger('test-launcher')
LAUNCHER_IFACE = "com.steampowered.PressureVessel.Launcher1"
LAUNCHER_PATH = "/com/steampowered/PressureVessel/Launcher1"
class TestLauncher(BaseTest):
def setUp(self) -> None:
super().setUp()
if 'PRESSURE_VESSEL_UNINSTALLED' in os.environ:
self.launcher = self.command_prefix + [
os.path.join(
self.top_builddir,
'pressure-vessel',
'pressure-vessel-launcher'
),
]
self.launch = self.command_prefix + [
os.path.join(
self.top_builddir,
'pressure-vessel',
'pressure-vessel-launch',
),
]
else:
self.skipTest('Not available as an installed-test')
def test_socket_directory(self) -> None:
with tempfile.TemporaryDirectory(prefix='test-') as temp:
need_terminate = True
printf_symlink = os.path.join(temp, 'printf=symlink')
printf = shutil.which('printf')
assert printf is not None
os.symlink(printf, printf_symlink)
proc = subprocess.Popen(
[
'env',
'PV_TEST_VAR=from-launcher',
] + self.launcher + [
'--socket-directory', temp,
],
stdout=subprocess.PIPE,
stderr=2,
universal_newlines=True,
)
try:
socket = ''
dbus_address = ''
stdout = proc.stdout
assert stdout is not None
for line in stdout:
line = line.rstrip('\n')
logger.debug('%s', line)
if line.startswith('socket='):
socket = line[len('socket='):]
elif line.startswith('dbus_address='):
dbus_address = (
line[len('dbus_address='):]
)
self.assertTrue(socket)
self.assertTrue(dbus_address)
left = os.stat(socket)
right = os.stat(os.path.join(temp, os.path.basename(socket)))
self.assertEqual(left.st_dev, right.st_dev)
self.assertEqual(left.st_ino, right.st_ino)
completed = run_subprocess(
self.launch + [
'--socket', socket,
'--',
'true',
],
check=True,
stdout=subprocess.PIPE,
stderr=2,
)
self.assertEqual(completed.stdout, b'')
completed = run_subprocess(
self.launch + [
'--socket', socket,
'--',
'printf', 'hello',
],
check=True,
stdout=subprocess.PIPE,
stderr=2,
)
self.assertEqual(completed.stdout, b'hello')
completed = run_subprocess(
self.launch + [
'--dbus-address', dbus_address,
'--',
'sh', '-euc', 'printf hello; printf W"ORL"D >&2',
],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
self.assertEqual(completed.stdout, b'hello')
self.assertIn(b'WORLD', completed.stderr)
completed = run_subprocess(
[
'env',
'PV_TEST_VAR=not-inherited',
] + self.launch + [
'--socket', socket,
'--',
'sh', '-euc', 'printf \'%s\' "$PV_TEST_VAR"',
],
check=True,
stdout=subprocess.PIPE,
stderr=2,
)
self.assertEqual(completed.stdout, b'from-launcher')
completed = run_subprocess(
[
'env',
'-u', 'PV_TEST_VAR',
'PV_TEST_VAR_ONE=one',
'PV_TEST_VAR_TWO=two',
] + self.launch + [
'--pass-env=PV_TEST_VAR',
'--pass-env-matching=PV_TEST_VAR_*',
'--socket', socket,
'--',
'sh', '-euc',
'printf \'%s\' '
'"'
'${PV_TEST_VAR-cleared}'
'${PV_TEST_VAR_ONE}'
'${PV_TEST_VAR_TWO}'
'"',
],
check=True,
stdout=subprocess.PIPE,
stderr=2,
)
self.assertEqual(completed.stdout, b'clearedonetwo')
completed = run_subprocess(
[
'env', '-u', 'PV_TEST_VAR',
] + self.launch + [
'--env=PV_TEST_VAR=nope',
'--pass-env=PV_TEST_VAR',
'--socket', socket,
'--',
'sh', '-euc', 'printf \'%s\' "${PV_TEST_VAR-cleared}"',
],
check=True,
stdout=subprocess.PIPE,
stderr=2,
)
self.assertEqual(completed.stdout, b'cleared')
completed = run_subprocess(
self.launch + [
'--env=PV_TEST_VAR=nope',
'--unset-env=PV_TEST_VAR',
'--socket', socket,
'--',
'sh', '-euc', 'printf \'%s\' "${PV_TEST_VAR-cleared}"',
],
check=True,
stdout=subprocess.PIPE,
stderr=2,
)
self.assertEqual(completed.stdout, b'cleared')
completed = run_subprocess(
self.launch + [
'--unset-env=PV_TEST_VAR',
'--socket', socket,
'--',
printf_symlink, 'hello',
],
check=True,
stdout=subprocess.PIPE,
stderr=2,
)
self.assertEqual(completed.stdout, b'hello')
completed = run_subprocess(
self.launch + [
'--clear-env',
'--socket', socket,
'--',
'sh', '-euc', 'echo -n "${PV_TEST_VAR-cleared}"',
],
check=True,
stdout=subprocess.PIPE,
stderr=2,
)
self.assertEqual(completed.stdout, b'cleared')
completed = run_subprocess(
[
'env',
'PV_TEST_VAR=not-inherited',
] + self.launch + [
'--pass-env=PV_TEST_VAR',
'--env=PV_TEST_VAR=nope',
'--env=PV_TEST_VAR=overridden',
'--socket', socket,
'--',
'sh', '-euc', 'printf \'%s\' "$PV_TEST_VAR"',
],
check=True,
stdout=subprocess.PIPE,
stderr=2,
)
self.assertEqual(completed.stdout, b'overridden')
launch = subprocess.Popen(
self.launch + [
'--socket', socket,
'--',
'sleep', '600',
],
stdout=subprocess.PIPE,
stderr=2,
)
launch.send_signal(signal.SIGINT)
self.assertIn(
launch.wait(),
(128 + signal.SIGINT, -signal.SIGINT),
)
completed = run_subprocess(
self.launch + [
'--socket', socket,
'--terminate',
'--',
'sh', '-euc', 'echo Goodbye',
],
check=True,
stdout=subprocess.PIPE,
stderr=2,
)
need_terminate = False
self.assertEqual(completed.stdout, b'Goodbye\n')
finally:
if need_terminate:
proc.terminate()
proc.wait()
self.assertEqual(proc.returncode, 0)
completed = run_subprocess(
self.launch + [
'--dbus-address', dbus_address,
'--',
'printf', 'hello',
],
stdout=2,
stderr=2,
)
self.assertEqual(completed.returncode, 125)
def test_socket(self) -> None:
with tempfile.TemporaryDirectory(prefix='test-') as temp:
need_terminate = True
proc = subprocess.Popen(
self.launcher + [
'--socket', os.path.join(temp, 'socket'),
],
stdout=subprocess.PIPE,
stderr=2,
universal_newlines=True,
)
try:
socket = ''
dbus_address = ''
stdout = proc.stdout
assert stdout is not None
for line in stdout:
line = line.rstrip('\n')
logger.debug('%s', line)
if line.startswith('socket='):
socket = line[len('socket='):]
elif line.startswith('dbus_address='):
dbus_address = (
line[len('dbus_address='):]
)
self.assertTrue(socket)
self.assertTrue(dbus_address)
# The path has been canonicalized, so it might not
# be equal to the input, but the basename will be the same
self.assertEqual(os.path.basename(socket), 'socket')
left = os.stat(socket)
right = os.stat(os.path.join(temp, 'socket'))
self.assertEqual(left.st_dev, right.st_dev)
self.assertEqual(left.st_ino, right.st_ino)
completed = run_subprocess(
self.launch + [
'--socket', socket,
'--',
'printf', 'hello',
],
check=True,
stdout=subprocess.PIPE,
stderr=2,
)
self.assertEqual(completed.stdout, b'hello')
completed = run_subprocess(
self.launch + [
'--dbus-address', dbus_address,
'--',
'sh', '-euc', 'printf hello; printf W"ORL"D >&2',
],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
self.assertEqual(completed.stdout, b'hello')
self.assertIn(b'WORLD', completed.stderr)
with self.subTest('direct D-Bus connection'):
try:
from gi.repository import GLib, Gio
except ImportError as e:
self.skipTest(str(e))
conn = Gio.DBusConnection.new_for_address_sync(
dbus_address,
Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT,
None,
None,
)
conn2 = Gio.DBusConnection.new_for_address_sync(
dbus_address,
Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT,
None,
None,
)
reply, out_fd_list = conn.call_with_unix_fd_list_sync(
None,
LAUNCHER_PATH,
LAUNCHER_IFACE,
'Launch',
GLib.Variant(
'(ayaaya{uh}a{ss}ua{sv})',
(
os.getcwdb() + b'\0',
[b'sleep\0', b'3600\0'],
{}, # no fds
{}, # environment
0, # no flags
{}, # no options
),
),
GLib.VariantType('(u)'),
Gio.DBusCallFlags.NONE,
-1,
None, # GUnixFDList is not bound
None,
)
if out_fd_list is not None:
self.assertEqual(out_fd_list.get_length(), 0)
self.assertEqual(len(reply), 1)
pid = int(reply[0])
os.kill(pid, 0)
with self.assertRaises(Exception) as catcher:
conn2.call_sync(
None,
LAUNCHER_PATH,
LAUNCHER_IFACE,
'SendSignal',
GLib.Variant('(uub)', (pid, signal.SIGTERM, True)),
GLib.VariantType('()'),
Gio.DBusCallFlags.NONE,
-1,
None,
)
logger.debug(
'When trying to kill process from wrong '
'connection: %s',
catcher.exception,
)
# still not dead
os.kill(pid, 0)
conn.call_sync(
None,
LAUNCHER_PATH,
LAUNCHER_IFACE,
'SendSignal',
GLib.Variant('(uub)', (pid, signal.SIGTERM, True)),
GLib.VariantType('()'),
Gio.DBusCallFlags.NONE,
-1,
None,
)
# This is stubbed out, because the pid could conceivably
# get reused for an unrelated process that doesn't exit -
# but in practice it does work
if 'PRESSURE_VESSEL_TEST_NO_STRAY_PROCESSES' in os.environ:
for i in range(20):
try:
os.kill(pid, 0)
except OSError:
break
else:
logger.debug('Process %d still alive', pid)
time.sleep(0.5)
continue
else:
raise AssertionError(
'Process %d did not exit' % pid
)
conn.call_sync(
None,
LAUNCHER_PATH,
LAUNCHER_IFACE,
'Terminate',
GLib.Variant('()', ()),
GLib.VariantType('()'),
Gio.DBusCallFlags.NONE,
-1,
None,
)
need_terminate = False
finally:
if need_terminate:
proc.terminate()
proc.wait()
self.assertEqual(proc.returncode, 0)
def test_session_bus(self) -> None:
try:
run_subprocess(
[
'dbus-send',
'--session',
'--dest=org.freedesktop.DBus',
'--print-reply',
'/org/freedesktop/DBus',
'org.freedesktop.DBus.Peer.Ping',
],
check=True,
)
except Exception:
self.skipTest('D-Bus session bus not available')
unique = '_' + uuid.uuid4().hex
proc = subprocess.Popen(
self.launcher + [
'--bus-name', 'com.steampowered.PressureVessel.Test.' + unique,
],
stdout=subprocess.PIPE,
stderr=2,
universal_newlines=True,
)
try:
bus_name = ''
stdout = proc.stdout
assert stdout is not None
for line in stdout:
line = line.rstrip('\n')
logger.debug('%s', line)
if line.startswith('bus_name='):
bus_name = line[len('bus_name='):]
self.assertEqual(
bus_name,
'com.steampowered.PressureVessel.Test.' + unique,
)
run_subprocess(
[
'dbus-send',
'--session',
'--dest=' + bus_name,
'--print-reply',
'/',
'org.freedesktop.DBus.Peer.Ping',
],
check=True,
stdout=subprocess.DEVNULL,
stderr=2,
)
completed = run_subprocess(
self.launch + [
'--bus-name', bus_name,
'--',
'printf', 'hello',
],
check=True,
stdout=subprocess.PIPE,
stderr=2,
)
self.assertEqual(completed.stdout, b'hello')
finally:
proc.terminate()
proc.wait()
self.assertEqual(proc.returncode, 0)
def test_exit_on_readable(self, use_stdin=False) -> None:
with tempfile.TemporaryDirectory(prefix='test-') as temp:
if not use_stdin:
read_end, write_end = os.pipe2(os.O_CLOEXEC)
pass_fds = [read_end]
fd = read_end
else:
pass_fds = []
fd = 0
read_end = 0
proc = subprocess.Popen(
self.launcher + [
'--socket', os.path.join(temp, 'socket'),
'--exit-on-readable=%d' % fd,
],
pass_fds=pass_fds,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=2,
universal_newlines=True,
)
stdin = proc.stdin
assert stdin is not None
os.close(read_end)
if use_stdin:
stdin.close()
else:
os.close(write_end)
proc.wait()
self.assertEqual(proc.returncode, 0)
def test_exit_on_stdin(self) -> None:
self.test_exit_on_readable(use_stdin=True)
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: