[lttng-tools.git] / tests / utils / lttngtest / environment.py

#!/usr/bin/env python3
#
# Copyright (C) 2022 Jérémie Galarneau <jeremie.galarneau@efficios.com>
#
# SPDX-License-Identifier: GPL-2.0-only
#

from types import FrameType
from typing import Callable, Iterator, Optional, Tuple, List
import sys
import pathlib
import signal
import subprocess
import shlex
import shutil
import os
import queue
import tempfile
from . import logger
import time
import threading
import contextlib


class TemporaryDirectory:
    def __init__(self, prefix):
        # type: (str) -> None
        self._directory_path = tempfile.mkdtemp(prefix=prefix)

    def __del__(self):
        shutil.rmtree(self._directory_path, ignore_errors=True)

    @property
    def path(self):
        # type: () -> pathlib.Path
        return pathlib.Path(self._directory_path)


class _SignalWaitQueue:
    """
    Utility class useful to wait for a signal before proceeding.

    Simply register the `signal` method as the handler for the signal you are
    interested in and call `wait_for_signal` to wait for its reception.

    Registering a signal:
        signal.signal(signal.SIGWHATEVER, queue.signal)

    Waiting for the signal:
        queue.wait_for_signal()
    """

    def __init__(self):
        self._queue = queue.Queue()  # type: queue.Queue

    def signal(
        self,
        signal_number,
        frame,  # type: Optional[FrameType]
    ):
        self._queue.put_nowait(signal_number)

    def wait_for_signal(self):
        self._queue.get(block=True)


class WaitTraceTestApplication:
    """
    Create an application that waits before tracing. This allows a test to
    launch an application, get its PID, and get it to start tracing when it
    has completed its setup.
    """

    def __init__(
        self,
        binary_path,  # type: pathlib.Path
        event_count,  # type: int
        environment,  # type: Environment
        wait_time_between_events_us=0,  # type: int
    ):
        self._environment = environment  # type: Environment
        if event_count % 5:
            # The test application currently produces 5 different events per iteration.
            raise ValueError("event count must be a multiple of 5")
        self._iteration_count = int(event_count / 5)  # type: int
        # File that the application will wait to see before tracing its events.
        self._app_start_tracing_file_path = pathlib.Path(
            tempfile.mktemp(
                prefix="app_",
                suffix="_start_tracing",
                dir=self._compat_open_path(environment.lttng_home_location),
            )
        )
        self._has_returned = False

        test_app_env = os.environ.copy()
        test_app_env["LTTNG_HOME"] = str(environment.lttng_home_location)
        # Make sure the app is blocked until it is properly registered to
        # the session daemon.
        test_app_env["LTTNG_UST_REGISTER_TIMEOUT"] = "-1"

        # File that the application will create to indicate it has completed its initialization.
        app_ready_file_path = tempfile.mktemp(
            prefix="app_",
            suffix="_ready",
            dir=self._compat_open_path(environment.lttng_home_location),
        )  # type: str

        test_app_args = [str(binary_path)]
        test_app_args.extend(
            shlex.split(
                "--iter {iteration_count} --create-in-main {app_ready_file_path} --wait-before-first-event {app_start_tracing_file_path} --wait {wait_time_between_events_us}".format(
                    iteration_count=self._iteration_count,
                    app_ready_file_path=app_ready_file_path,
                    app_start_tracing_file_path=self._app_start_tracing_file_path,
                    wait_time_between_events_us=wait_time_between_events_us,
                )
            )
        )

        self._process = subprocess.Popen(
            test_app_args,
            env=test_app_env,
        )  # type: subprocess.Popen

        # Wait for the application to create the file indicating it has fully
        # initialized. Make sure the app hasn't crashed in order to not wait
        # forever.
        while True:
            if os.path.exists(app_ready_file_path):
                break

            if self._process.poll() is not None:
                # Application has unexepectedly returned.
                raise RuntimeError(
                    "Test application has unexepectedly returned during its initialization with return code `{return_code}`".format(
                        return_code=self._process.returncode
                    )
                )

            time.sleep(0.1)

    def trace(self):
        # type: () -> None
        if self._process.poll() is not None:
            # Application has unexepectedly returned.
            raise RuntimeError(
                "Test application has unexepectedly before tracing with return code `{return_code}`".format(
                    return_code=self._process.returncode
                )
            )
        open(self._compat_open_path(self._app_start_tracing_file_path), mode="x")

    def wait_for_exit(self):
        # type: () -> None
        if self._process.wait() != 0:
            raise RuntimeError(
                "Test application has exit with return code `{return_code}`".format(
                    return_code=self._process.returncode
                )
            )
        self._has_returned = True

    @property
    def vpid(self):
        # type: () -> int
        return self._process.pid

    @staticmethod
    def _compat_open_path(path):
        # type: (pathlib.Path) -> pathlib.Path | str
        """
        The builtin open() in python >= 3.6 expects a path-like object while
        prior versions expect a string or bytes object. Return the correct type
        based on the presence of the "__fspath__" attribute specified in PEP-519.
        """
        if hasattr(path, "__fspath__"):
            return path
        else:
            return str(path)

    def __del__(self):
        if not self._has_returned:
            # This is potentially racy if the pid has been recycled. However,
            # we can't use pidfd_open since it is only available in python >= 3.9.
            self._process.kill()
            self._process.wait()


class TraceTestApplication:
    """
    Create an application to trace.
    """

    def __init__(self, binary_path: pathlib.Path, environment: "Environment"):
        self._environment: Environment = environment
        self._has_returned = False

        test_app_env = os.environ.copy()
        test_app_env["LTTNG_HOME"] = str(environment.lttng_home_location)
        # Make sure the app is blocked until it is properly registered to
        # the session daemon.
        test_app_env["LTTNG_UST_REGISTER_TIMEOUT"] = "-1"

        test_app_args = [str(binary_path)]

        self._process: subprocess.Popen = subprocess.Popen(
            test_app_args, env=test_app_env
        )

    def wait_for_exit(self) -> None:
        if self._process.wait() != 0:
            raise RuntimeError(
                "Test application has exit with return code `{return_code}`".format(
                    return_code=self._process.returncode
                )
            )
        self._has_returned = True

    def __del__(self):
        if not self._has_returned:
            # This is potentially racy if the pid has been recycled. However,
            # we can't use pidfd_open since it is only available in python >= 3.9.
            self._process.kill()
            self._process.wait()


class ProcessOutputConsumer(threading.Thread, logger._Logger):
    def __init__(
        self,
        process,  # type: subprocess.Popen
        name,  # type: str
        log,  # type: Callable[[str], None]
    ):
        threading.Thread.__init__(self)
        self._prefix = name
        logger._Logger.__init__(self, log)
        self._process = process

    def run(self):
        # type: () -> None
        while self._process.poll() is None:
            assert self._process.stdout
            line = self._process.stdout.readline().decode("utf-8").replace("\n", "")
            if len(line) != 0:
                self._log("{prefix}: {line}".format(prefix=self._prefix, line=line))


# Generate a temporary environment in which to execute a test.
class _Environment(logger._Logger):
    def __init__(
        self,
        with_sessiond,  # type: bool
        log=None,  # type: Optional[Callable[[str], None]]
    ):
        super().__init__(log)
        signal.signal(signal.SIGTERM, self._handle_termination_signal)
        signal.signal(signal.SIGINT, self._handle_termination_signal)

        # Assumes the project's hierarchy to this file is:
        # tests/utils/python/this_file
        self._project_root = (
            pathlib.Path(__file__).absolute().parents[3]
        )  # type: pathlib.Path
        self._lttng_home = TemporaryDirectory(
            "lttng_test_env_home"
        )  # type: Optional[TemporaryDirectory]

        self._sessiond = (
            self._launch_lttng_sessiond() if with_sessiond else None
        )  # type: Optional[subprocess.Popen[bytes]]

    @property
    def lttng_home_location(self):
        # type: () -> pathlib.Path
        if self._lttng_home is None:
            raise RuntimeError("Attempt to access LTTng home after clean-up")
        return self._lttng_home.path

    @property
    def lttng_client_path(self):
        # type: () -> pathlib.Path
        return self._project_root / "src" / "bin" / "lttng" / "lttng"

    def create_temporary_directory(self, prefix=None):
        # type: (Optional[str]) -> pathlib.Path
        # Simply return a path that is contained within LTTNG_HOME; it will
        # be destroyed when the temporary home goes out of scope.
        assert self._lttng_home
        return pathlib.Path(
            tempfile.mkdtemp(
                prefix="tmp" if prefix is None else prefix,
                dir=str(self._lttng_home.path),
            )
        )

    # Unpack a list of environment variables from a string
    # such as "HELLO=is_it ME='/you/are/looking/for'"
    @staticmethod
    def _unpack_env_vars(env_vars_string):
        # type: (str) -> List[Tuple[str, str]]
        unpacked_vars = []
        for var in shlex.split(env_vars_string):
            equal_position = var.find("=")
            # Must have an equal sign and not end with an equal sign
            if equal_position == -1 or equal_position == len(var) - 1:
                raise ValueError(
                    "Invalid sessiond environment variable: `{}`".format(var)
                )

            var_name = var[0:equal_position]
            var_value = var[equal_position + 1 :]
            # Unquote any paths
            var_value = var_value.replace("'", "")
            var_value = var_value.replace('"', "")
            unpacked_vars.append((var_name, var_value))

        return unpacked_vars

    def _launch_lttng_sessiond(self):
        # type: () -> Optional[subprocess.Popen]
        is_64bits_host = sys.maxsize > 2**32

        sessiond_path = (
            self._project_root / "src" / "bin" / "lttng-sessiond" / "lttng-sessiond"
        )
        consumerd_path_option_name = "--consumerd{bitness}-path".format(
            bitness="64" if is_64bits_host else "32"
        )
        consumerd_path = (
            self._project_root / "src" / "bin" / "lttng-consumerd" / "lttng-consumerd"
        )

        no_sessiond_var = os.environ.get("TEST_NO_SESSIOND")
        if no_sessiond_var and no_sessiond_var == "1":
            # Run test without a session daemon; the user probably
            # intends to run one under gdb for example.
            return None

        # Setup the session daemon's environment
        sessiond_env_vars = os.environ.get("LTTNG_SESSIOND_ENV_VARS")
        sessiond_env = os.environ.copy()
        if sessiond_env_vars:
            self._log("Additional lttng-sessiond environment variables:")
            additional_vars = self._unpack_env_vars(sessiond_env_vars)
            for var_name, var_value in additional_vars:
                self._log("  {name}={value}".format(name=var_name, value=var_value))
                sessiond_env[var_name] = var_value

        sessiond_env["LTTNG_SESSION_CONFIG_XSD_PATH"] = str(
            self._project_root / "src" / "common"
        )

        assert self._lttng_home is not None
        sessiond_env["LTTNG_HOME"] = str(self._lttng_home.path)

        wait_queue = _SignalWaitQueue()
        signal.signal(signal.SIGUSR1, wait_queue.signal)

        self._log(
            "Launching session daemon with LTTNG_HOME=`{home_dir}`".format(
                home_dir=str(self._lttng_home.path)
            )
        )
        process = subprocess.Popen(
            [
                str(sessiond_path),
                consumerd_path_option_name,
                str(consumerd_path),
                "--sig-parent",
            ],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            env=sessiond_env,
        )

        if self._logging_function:
            self._sessiond_output_consumer = ProcessOutputConsumer(
                process, "lttng-sessiond", self._logging_function
            )  # type: Optional[ProcessOutputConsumer]
            self._sessiond_output_consumer.daemon = True
            self._sessiond_output_consumer.start()

        # Wait for SIGUSR1, indicating the sessiond is ready to proceed
        wait_queue.wait_for_signal()
        signal.signal(signal.SIGUSR1, wait_queue.signal)

        return process

    def _handle_termination_signal(self, signal_number, frame):
        # type: (int, Optional[FrameType]) -> None
        self._log(
            "Killed by {signal_name} signal, cleaning-up".format(
                signal_name=signal.strsignal(signal_number)
            )
        )
        self._cleanup()

    def launch_wait_trace_test_application(self, event_count):
        # type: (int) -> WaitTraceTestApplication
        """
        Launch an application that will wait before tracing `event_count` events.
        """
        return WaitTraceTestApplication(
            self._project_root
            / "tests"
            / "utils"
            / "testapp"
            / "gen-ust-nevents"
            / "gen-ust-nevents",
            event_count,
            self,
        )

    def launch_trace_test_constructor_application(
        self
    ) -> TraceTestApplication:
        """
        Launch an application that will trace from within constructors.
        """
        return TraceTestApplication(
            self._project_root
            / "tests"
            / "utils"
            / "testapp"
            / "gen-ust-events-constructor"
            / "gen-ust-events-constructor",
            self,
        )

    # Clean-up managed processes
    def _cleanup(self):
        # type: () -> None
        if self._sessiond and self._sessiond.poll() is None:
            # The session daemon is alive; kill it.
            self._log(
                "Killing session daemon (pid = {sessiond_pid})".format(
                    sessiond_pid=self._sessiond.pid
                )
            )

            self._sessiond.terminate()
            self._sessiond.wait()
            if self._sessiond_output_consumer:
                self._sessiond_output_consumer.join()
                self._sessiond_output_consumer = None

            self._log("Session daemon killed")
            self._sessiond = None

        self._lttng_home = None

    def __del__(self):
        self._cleanup()


@contextlib.contextmanager
def test_environment(with_sessiond, log=None):
    # type: (bool, Optional[Callable[[str], None]]) -> Iterator[_Environment]
    env = _Environment(with_sessiond, log)
    try:
        yield env
    finally:
        env._cleanup()
Commit	Line	Data
ef945e4d JG	1	#!/usr/bin/env python3
	2	#
	3	# Copyright (C) 2022 Jérémie Galarneau <jeremie.galarneau@efficios.com>
	4	#
	5	# SPDX-License-Identifier: GPL-2.0-only
	6	#
	7
	8	from types import FrameType
ce8470c9	9	from typing import Callable, Iterator, Optional, Tuple, List
ef945e4d JG	10	import sys
	11	import pathlib
	12	import signal
	13	import subprocess
	14	import shlex
	15	import shutil
	16	import os
	17	import queue
	18	import tempfile
	19	from . import logger
	20	import time
	21	import threading
	22	import contextlib
	23
	24
	25	class TemporaryDirectory:
ce8470c9 MJ	26	def __init__(self, prefix):
ce8470c9 MJ	27	# type: (str) -> None
ef945e4d JG	28	self._directory_path = tempfile.mkdtemp(prefix=prefix)
	29
	30	def __del__(self):
	31	shutil.rmtree(self._directory_path, ignore_errors=True)
	32
	33	@property
ce8470c9 MJ	34	def path(self):
ce8470c9 MJ	35	# type: () -> pathlib.Path
ef945e4d JG	36	return pathlib.Path(self._directory_path)
	37
	38
	39	class _SignalWaitQueue:
	40	"""
	41	Utility class useful to wait for a signal before proceeding.
	42
	43	Simply register the `signal` method as the handler for the signal you are
	44	interested in and call `wait_for_signal` to wait for its reception.
	45
	46	Registering a signal:
	47	signal.signal(signal.SIGWHATEVER, queue.signal)
	48
	49	Waiting for the signal:
	50	queue.wait_for_signal()
	51	"""
	52
	53	def __init__(self):
ce8470c9	54	self._queue = queue.Queue() # type: queue.Queue
ef945e4d	55
ce8470c9 MJ	56	def signal(
	57	self,
	58	signal_number,
	59	frame, # type: Optional[FrameType]
	60	):
ef945e4d JG	61	self._queue.put_nowait(signal_number)
	62
	63	def wait_for_signal(self):
	64	self._queue.get(block=True)
	65
	66
	67	class WaitTraceTestApplication:
	68	"""
	69	Create an application that waits before tracing. This allows a test to
	70	launch an application, get its PID, and get it to start tracing when it
	71	has completed its setup.
	72	"""
	73
	74	def __init__(
	75	self,
ce8470c9 MJ	76	binary_path, # type: pathlib.Path
	77	event_count, # type: int
	78	environment, # type: Environment
	79	wait_time_between_events_us=0, # type: int
ef945e4d	80	):
ce8470c9	81	self._environment = environment # type: Environment
ef945e4d JG	82	if event_count % 5:
	83	# The test application currently produces 5 different events per iteration.
	84	raise ValueError("event count must be a multiple of 5")
ce8470c9	85	self._iteration_count = int(event_count / 5) # type: int
ef945e4d	86	# File that the application will wait to see before tracing its events.
ce8470c9	87	self._app_start_tracing_file_path = pathlib.Path(
ef945e4d JG	88	tempfile.mktemp(
	89	prefix="app_",
	90	suffix="_start_tracing",
2d2198ca	91	dir=self._compat_open_path(environment.lttng_home_location),
ef945e4d JG	92	)
	93	)
	94	self._has_returned = False
	95
	96	test_app_env = os.environ.copy()
	97	test_app_env["LTTNG_HOME"] = str(environment.lttng_home_location)
	98	# Make sure the app is blocked until it is properly registered to
	99	# the session daemon.
	100	test_app_env["LTTNG_UST_REGISTER_TIMEOUT"] = "-1"
	101
	102	# File that the application will create to indicate it has completed its initialization.
8466f071	103	app_ready_file_path = tempfile.mktemp(
2d2198ca MJ	104	prefix="app_",
	105	suffix="_ready",
	106	dir=self._compat_open_path(environment.lttng_home_location),
ce8470c9	107	) # type: str
ef945e4d JG	108
	109	test_app_args = [str(binary_path)]
	110	test_app_args.extend(
	111	shlex.split(
	112	"--iter {iteration_count} --create-in-main {app_ready_file_path} --wait-before-first-event {app_start_tracing_file_path} --wait {wait_time_between_events_us}".format(
	113	iteration_count=self._iteration_count,
	114	app_ready_file_path=app_ready_file_path,
	115	app_start_tracing_file_path=self._app_start_tracing_file_path,
	116	wait_time_between_events_us=wait_time_between_events_us,
	117	)
	118	)
	119	)
	120
ce8470c9	121	self._process = subprocess.Popen(
ef945e4d JG	122	test_app_args,
ef945e4d JG	123	env=test_app_env,
ce8470c9	124	) # type: subprocess.Popen
ef945e4d JG	125
	126	# Wait for the application to create the file indicating it has fully
	127	# initialized. Make sure the app hasn't crashed in order to not wait
	128	# forever.
	129	while True:
	130	if os.path.exists(app_ready_file_path):
	131	break
	132
	133	if self._process.poll() is not None:
	134	# Application has unexepectedly returned.
	135	raise RuntimeError(
	136	"Test application has unexepectedly returned during its initialization with return code `{return_code}`".format(
	137	return_code=self._process.returncode
	138	)
	139	)
	140
	141	time.sleep(0.1)
	142
ce8470c9 MJ	143	def trace(self):
ce8470c9 MJ	144	# type: () -> None
ef945e4d JG	145	if self._process.poll() is not None:
	146	# Application has unexepectedly returned.
	147	raise RuntimeError(
	148	"Test application has unexepectedly before tracing with return code `{return_code}`".format(
	149	return_code=self._process.returncode
	150	)
	151	)
2d2198ca	152	open(self._compat_open_path(self._app_start_tracing_file_path), mode="x")
ef945e4d	153
ce8470c9 MJ	154	def wait_for_exit(self):
ce8470c9 MJ	155	# type: () -> None
ef945e4d JG	156	if self._process.wait() != 0:
	157	raise RuntimeError(
	158	"Test application has exit with return code `{return_code}`".format(
	159	return_code=self._process.returncode
	160	)
	161	)
	162	self._has_returned = True
	163
	164	@property
ce8470c9 MJ	165	def vpid(self):
ce8470c9 MJ	166	# type: () -> int
ef945e4d JG	167	return self._process.pid
ef945e4d JG	168
2d2198ca MJ	169	@staticmethod
2d2198ca MJ	170	def _compat_open_path(path):
ce8470c9	171	# type: (pathlib.Path) -> pathlib.Path \| str
2d2198ca MJ	172	"""
	173	The builtin open() in python >= 3.6 expects a path-like object while
	174	prior versions expect a string or bytes object. Return the correct type
	175	based on the presence of the "__fspath__" attribute specified in PEP-519.
	176	"""
	177	if hasattr(path, "__fspath__"):
	178	return path
	179	else:
	180	return str(path)
	181
ef945e4d JG	182	def __del__(self):
	183	if not self._has_returned:
	184	# This is potentially racy if the pid has been recycled. However,
	185	# we can't use pidfd_open since it is only available in python >= 3.9.
	186	self._process.kill()
	187	self._process.wait()
	188
	189
da1e97c9 MD	190	class TraceTestApplication:
	191	"""
	192	Create an application to trace.
	193	"""
	194
	195	def __init__(self, binary_path: pathlib.Path, environment: "Environment"):
	196	self._environment: Environment = environment
	197	self._has_returned = False
	198
	199	test_app_env = os.environ.copy()
	200	test_app_env["LTTNG_HOME"] = str(environment.lttng_home_location)
	201	# Make sure the app is blocked until it is properly registered to
	202	# the session daemon.
	203	test_app_env["LTTNG_UST_REGISTER_TIMEOUT"] = "-1"
	204
	205	test_app_args = [str(binary_path)]
	206
	207	self._process: subprocess.Popen = subprocess.Popen(
	208	test_app_args, env=test_app_env
	209	)
	210
	211	def wait_for_exit(self) -> None:
	212	if self._process.wait() != 0:
	213	raise RuntimeError(
	214	"Test application has exit with return code `{return_code}`".format(
	215	return_code=self._process.returncode
	216	)
	217	)
	218	self._has_returned = True
	219
	220	def __del__(self):
	221	if not self._has_returned:
	222	# This is potentially racy if the pid has been recycled. However,
	223	# we can't use pidfd_open since it is only available in python >= 3.9.
	224	self._process.kill()
	225	self._process.wait()
	226
	227
ef945e4d JG	228	class ProcessOutputConsumer(threading.Thread, logger._Logger):
ef945e4d JG	229	def __init__(
ce8470c9 MJ	230	self,
	231	process, # type: subprocess.Popen
	232	name, # type: str
	233	log, # type: Callable[[str], None]
ef945e4d JG	234	):
	235	threading.Thread.__init__(self)
	236	self._prefix = name
	237	logger._Logger.__init__(self, log)
	238	self._process = process
	239
ce8470c9 MJ	240	def run(self):
ce8470c9 MJ	241	# type: () -> None
ef945e4d JG	242	while self._process.poll() is None:
	243	assert self._process.stdout
	244	line = self._process.stdout.readline().decode("utf-8").replace("\n", "")
	245	if len(line) != 0:
	246	self._log("{prefix}: {line}".format(prefix=self._prefix, line=line))
	247
	248
	249	# Generate a temporary environment in which to execute a test.
	250	class _Environment(logger._Logger):
	251	def __init__(
ce8470c9 MJ	252	self,
	253	with_sessiond, # type: bool
	254	log=None, # type: Optional[Callable[[str], None]]
ef945e4d JG	255	):
	256	super().__init__(log)
	257	signal.signal(signal.SIGTERM, self._handle_termination_signal)
	258	signal.signal(signal.SIGINT, self._handle_termination_signal)
	259
	260	# Assumes the project's hierarchy to this file is:
	261	# tests/utils/python/this_file
ce8470c9 MJ	262	self._project_root = (
	263	pathlib.Path(__file__).absolute().parents[3]
	264	) # type: pathlib.Path
	265	self._lttng_home = TemporaryDirectory(
ef945e4d	266	"lttng_test_env_home"
ce8470c9	267	) # type: Optional[TemporaryDirectory]
ef945e4d	268
ce8470c9	269	self._sessiond = (
ef945e4d	270	self._launch_lttng_sessiond() if with_sessiond else None
ce8470c9	271	) # type: Optional[subprocess.Popen[bytes]]
ef945e4d JG	272
ef945e4d JG	273	@property
ce8470c9 MJ	274	def lttng_home_location(self):
ce8470c9 MJ	275	# type: () -> pathlib.Path
ef945e4d JG	276	if self._lttng_home is None:
	277	raise RuntimeError("Attempt to access LTTng home after clean-up")
	278	return self._lttng_home.path
	279
	280	@property
ce8470c9 MJ	281	def lttng_client_path(self):
ce8470c9 MJ	282	# type: () -> pathlib.Path
ef945e4d JG	283	return self._project_root / "src" / "bin" / "lttng" / "lttng"
ef945e4d JG	284
ce8470c9 MJ	285	def create_temporary_directory(self, prefix=None):
ce8470c9 MJ	286	# type: (Optional[str]) -> pathlib.Path
ef945e4d JG	287	# Simply return a path that is contained within LTTNG_HOME; it will
	288	# be destroyed when the temporary home goes out of scope.
	289	assert self._lttng_home
	290	return pathlib.Path(
	291	tempfile.mkdtemp(
	292	prefix="tmp" if prefix is None else prefix,
	293	dir=str(self._lttng_home.path),
	294	)
	295	)
	296
	297	# Unpack a list of environment variables from a string
	298	# such as "HELLO=is_it ME='/you/are/looking/for'"
	299	@staticmethod
ce8470c9 MJ	300	def _unpack_env_vars(env_vars_string):
ce8470c9 MJ	301	# type: (str) -> List[Tuple[str, str]]
ef945e4d JG	302	unpacked_vars = []
	303	for var in shlex.split(env_vars_string):
	304	equal_position = var.find("=")
	305	# Must have an equal sign and not end with an equal sign
	306	if equal_position == -1 or equal_position == len(var) - 1:
	307	raise ValueError(
	308	"Invalid sessiond environment variable: `{}`".format(var)
	309	)
	310
	311	var_name = var[0:equal_position]
	312	var_value = var[equal_position + 1 :]
	313	# Unquote any paths
	314	var_value = var_value.replace("'", "")
	315	var_value = var_value.replace('"', "")
	316	unpacked_vars.append((var_name, var_value))
	317
	318	return unpacked_vars
	319
ce8470c9 MJ	320	def _launch_lttng_sessiond(self):
ce8470c9 MJ	321	# type: () -> Optional[subprocess.Popen]
ef945e4d JG	322	is_64bits_host = sys.maxsize > 2**32
	323
	324	sessiond_path = (
	325	self._project_root / "src" / "bin" / "lttng-sessiond" / "lttng-sessiond"
	326	)
	327	consumerd_path_option_name = "--consumerd{bitness}-path".format(
	328	bitness="64" if is_64bits_host else "32"
	329	)
	330	consumerd_path = (
	331	self._project_root / "src" / "bin" / "lttng-consumerd" / "lttng-consumerd"
	332	)
	333
	334	no_sessiond_var = os.environ.get("TEST_NO_SESSIOND")
	335	if no_sessiond_var and no_sessiond_var == "1":
	336	# Run test without a session daemon; the user probably
	337	# intends to run one under gdb for example.
	338	return None
	339
	340	# Setup the session daemon's environment
	341	sessiond_env_vars = os.environ.get("LTTNG_SESSIOND_ENV_VARS")
	342	sessiond_env = os.environ.copy()
	343	if sessiond_env_vars:
	344	self._log("Additional lttng-sessiond environment variables:")
	345	additional_vars = self._unpack_env_vars(sessiond_env_vars)
	346	for var_name, var_value in additional_vars:
	347	self._log(" {name}={value}".format(name=var_name, value=var_value))
	348	sessiond_env[var_name] = var_value
	349
	350	sessiond_env["LTTNG_SESSION_CONFIG_XSD_PATH"] = str(
	351	self._project_root / "src" / "common"
	352	)
	353
	354	assert self._lttng_home is not None
	355	sessiond_env["LTTNG_HOME"] = str(self._lttng_home.path)
	356
	357	wait_queue = _SignalWaitQueue()
	358	signal.signal(signal.SIGUSR1, wait_queue.signal)
	359
	360	self._log(
	361	"Launching session daemon with LTTNG_HOME=`{home_dir}`".format(
	362	home_dir=str(self._lttng_home.path)
	363	)
	364	)
	365	process = subprocess.Popen(
	366	[
	367	str(sessiond_path),
	368	consumerd_path_option_name,
	369	str(consumerd_path),
	370	"--sig-parent",
	371	],
	372	stdout=subprocess.PIPE,
	373	stderr=subprocess.STDOUT,
	374	env=sessiond_env,
	375	)
	376
	377	if self._logging_function:
ce8470c9 MJ	378	self._sessiond_output_consumer = ProcessOutputConsumer(
	379	process, "lttng-sessiond", self._logging_function
	380	) # type: Optional[ProcessOutputConsumer]
ef945e4d JG	381	self._sessiond_output_consumer.daemon = True
	382	self._sessiond_output_consumer.start()
	383
	384	# Wait for SIGUSR1, indicating the sessiond is ready to proceed
	385	wait_queue.wait_for_signal()
	386	signal.signal(signal.SIGUSR1, wait_queue.signal)
	387
	388	return process
	389
ce8470c9 MJ	390	def _handle_termination_signal(self, signal_number, frame):
ce8470c9 MJ	391	# type: (int, Optional[FrameType]) -> None
ef945e4d JG	392	self._log(
	393	"Killed by {signal_name} signal, cleaning-up".format(
	394	signal_name=signal.strsignal(signal_number)
	395	)
	396	)
	397	self._cleanup()
	398
ce8470c9 MJ	399	def launch_wait_trace_test_application(self, event_count):
ce8470c9 MJ	400	# type: (int) -> WaitTraceTestApplication
ef945e4d JG	401	"""
	402	Launch an application that will wait before tracing `event_count` events.
	403	"""
	404	return WaitTraceTestApplication(
	405	self._project_root
	406	/ "tests"
	407	/ "utils"
	408	/ "testapp"
	409	/ "gen-ust-nevents"
	410	/ "gen-ust-nevents",
	411	event_count,
	412	self,
	413	)
	414
da1e97c9 MD	415	def launch_trace_test_constructor_application(
	416	self
	417	) -> TraceTestApplication:
	418	"""
	419	Launch an application that will trace from within constructors.
	420	"""
	421	return TraceTestApplication(
	422	self._project_root
	423	/ "tests"
	424	/ "utils"
	425	/ "testapp"
	426	/ "gen-ust-events-constructor"
	427	/ "gen-ust-events-constructor",
	428	self,
	429	)
	430
ef945e4d	431	# Clean-up managed processes
ce8470c9 MJ	432	def _cleanup(self):
ce8470c9 MJ	433	# type: () -> None
ef945e4d JG	434	if self._sessiond and self._sessiond.poll() is None:
	435	# The session daemon is alive; kill it.
	436	self._log(
	437	"Killing session daemon (pid = {sessiond_pid})".format(
	438	sessiond_pid=self._sessiond.pid
	439	)
	440	)
	441
	442	self._sessiond.terminate()
	443	self._sessiond.wait()
	444	if self._sessiond_output_consumer:
	445	self._sessiond_output_consumer.join()
	446	self._sessiond_output_consumer = None
	447
	448	self._log("Session daemon killed")
	449	self._sessiond = None
	450
	451	self._lttng_home = None
	452
	453	def __del__(self):
	454	self._cleanup()
	455
	456
	457	@contextlib.contextmanager
ce8470c9 MJ	458	def test_environment(with_sessiond, log=None):
ce8470c9 MJ	459	# type: (bool, Optional[Callable[[str], None]]) -> Iterator[_Environment]
ef945e4d JG	460	env = _Environment(with_sessiond, log)
	461	try:
	462	yield env
	463	finally:
	464	env._cleanup()