Class TestProcess

locust/test/subprocess_utils.py:14–183

Wraps a subprocess for testing purposes.

class TestProcess:
    """
    Wraps a subprocess for testing purposes.
    """

    __test__ = False

    def __init__(
        self,
        command: str,
        *,
        extra_env: dict[str, str] = {},
        expect_return_code: int | None = 0,
        sigint_on_exit: bool = True,
        expect_timeout: int = 5,
        use_pty: bool = False,
        join_timeout: int = 1,
    ):
        self.proc: subprocess.Popen[str]
        self._terminated = False
        self._failed = False

        self.expect_return_code = expect_return_code
        self.expect_timeout = expect_timeout
        self.sigint_on_exit = sigint_on_exit
        self.join_timeout = join_timeout

        self.stderr_output: list[str] = []
        self.stdout_output: list[str] = []
        self._stderr_cursor: int = 0  # Used for stateful log matching
        self._stdout_cursor: int = 0

        self.use_pty: bool = use_pty
        # Create PTY pair
        if self.use_pty:
            if IS_WINDOWS:
                raise Exception("termios doesn't exist on windows, and thus we cannot import pty")

            import pty

            self.stdin_m, self.stdin_s = pty.openpty()

        self.proc = subprocess.Popen(
            shlex.split(command) if not IS_WINDOWS else command.split(" "),
            env={
                "PYTHONUNBUFFERED": "1",
                **os.environ,
                "OTEL_LOGS_EXPORTER": "none",
                "OTEL_METRICS_EXPORTER": "none",
                "OTEL_TRACES_EXPORTER": "none",
                **extra_env,
            },
            stdin=self.stdin_s if self.use_pty else None,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )