GaudiTesting.SubprocessBaseTest.SubprocessBaseTest Class Reference

Inheritance diagram for GaudiTesting.SubprocessBaseTest.SubprocessBaseTest:

Public Member Functions
Path|None	cwd (self)
str	resolve_path (cls, Path|str path)
None	update_env (cls, dict[str, str] env)
FixtureResult	run_program (cls, tmp_path=None)
	run_program_for_dbg (cls, tmp_path)
None	test_fixture_setup (self, Callable[[str, str], None] record_property, FixtureResult fixture_result, Path|None reference_path)
None	test_returncode (self, int returncode)

Static Public Member Functions
str	expand_vars_from (str value, dict[str, str] env)
None	unset_vars (dict[str, str] env, list[str] vars_to_unset)

Static Public Attributes
list	command = None
str	reference = None
list	environment = None
int	timeout = 600
int	returncode = 0
dict	popen_kwargs = {}
list	env_to_record

Protected Member Functions
dict[str, str]	_prepare_environment (cls)
str	_determine_program (cls, str prog)
list[str]	_prepare_command (cls, tmp_path=Path())
str	_handle_timeout (cls, subprocess.Popen proc)
	_prepare_execution (cls, tmp_path=None)

Static Protected Member Functions
str	_collect_stack_trace (subprocess.Popen proc)
None	_terminate_process (subprocess.Popen proc)

Detailed Description

A base class for running and managing subprocess executions within a test framework.
It provides mechanisms for setting up the environment, preparing commands,
and handling subprocess output and errors.

Definition at line 35 of file SubprocessBaseTest.py.

Member Function Documentation

_collect_stack_trace()

str GaudiTesting.SubprocessBaseTest.SubprocessBaseTest._collect_stack_trace ( subprocess.Popen proc )

staticprotected

Collect stack trace from a running process using the available debugger.
Uses lldb on macOS and gdb on Linux. Returns a message if no debugger
is available.

Definition at line 134 of file SubprocessBaseTest.py.

    def _collect_stack_trace(proc: subprocess.Popen) -> str:
        """
        Collect stack trace from a running process using the available debugger.
        Uses lldb on macOS and gdb on Linux. Returns a message if no debugger
        is available.
        """
        import sys
 
        if sys.platform == "darwin":
            # Use lldb on macOS
            if not shutil.which("lldb"):
                return "(lldb not available for stack trace collection)"
            cmd = [
                "lldb",
                "-p",
                str(proc.pid),
                "-o",
                "bt all",
                "-o",
                "quit",
            ]
        else:
            # Use gdb on Linux
            if not shutil.which("gdb"):
                return "(gdb not available for stack trace collection)"
            cmd = [
                "gdb",
                "--pid",
                str(proc.pid),
                "--batch",
                "--eval-command=thread apply all backtrace",
            ]
        debugger = subprocess.Popen(
            cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
        )
        return debugger.communicate()[0].decode("utf-8", errors="backslashreplace")
 

_determine_program()

str GaudiTesting.SubprocessBaseTest.SubprocessBaseTest._determine_program ( cls, str prog )

protected

Definition at line 106 of file SubprocessBaseTest.py.

    def _determine_program(cls, prog: str) -> str:
        if not any(prog.lower().endswith(ext) for ext in [".exe", ".py", ".bat"]):
            prog += ".exe"
        return which(prog) or cls.resolve_path(prog)
 

_handle_timeout()

str GaudiTesting.SubprocessBaseTest.SubprocessBaseTest._handle_timeout ( cls, subprocess.Popen proc )

protected

Handle a process timeout by collecting and returning the stack trace.

Definition at line 125 of file SubprocessBaseTest.py.

    def _handle_timeout(cls, proc: subprocess.Popen) -> str:
        """
        Handle a process timeout by collecting and returning the stack trace.
        """
        stack_trace = cls._collect_stack_trace(proc)
        cls._terminate_process(proc)
        return stack_trace
 

_prepare_command()

list[str] GaudiTesting.SubprocessBaseTest.SubprocessBaseTest._prepare_command ( cls, tmp_path = Path() )

protected

Prepare the command to be executed, resolving paths for each part.

Reimplemented in GaudiTesting.GaudiExeTest.GaudiExeTest.

Definition at line 112 of file SubprocessBaseTest.py.

    def _prepare_command(cls, tmp_path=Path()) -> list[str]:
        """
        Prepare the command to be executed, resolving paths for each part.
        """
        command = [cls._determine_program(cls.command[0])]
        for part in cls.command[1:]:
            if not part.startswith("-"):  # do not try to expand options
                command.append(cls.resolve_path(part))
            else:
                command.append(part)
        return command
 

_prepare_environment()

dict[str, str] GaudiTesting.SubprocessBaseTest.SubprocessBaseTest._prepare_environment ( cls )

protected

Definition at line 91 of file SubprocessBaseTest.py.

    def _prepare_environment(cls) -> dict[str, str]:
        env = dict(os.environ)
        cls.update_env(env)
        return env
 

_prepare_execution()

GaudiTesting.SubprocessBaseTest.SubprocessBaseTest._prepare_execution ( cls, tmp_path = None )

protected

Definition at line 179 of file SubprocessBaseTest.py.

    def _prepare_execution(cls, tmp_path=None):
        command = cls._prepare_command(tmp_path=tmp_path)
        env = cls._prepare_environment()
        os.makedirs(cls.popen_kwargs.get("cwd", tmp_path), exist_ok=True)
        return command, env
 

_terminate_process()

None GaudiTesting.SubprocessBaseTest.SubprocessBaseTest._terminate_process ( subprocess.Popen proc )

staticprotected

Definition at line 172 of file SubprocessBaseTest.py.

    def _terminate_process(proc: subprocess.Popen) -> None:
        kill_tree(proc.pid, signal.SIGTERM)
        proc.wait(60)
        if proc.poll() is None:
            kill_tree(proc.pid, signal.SIGKILL)
 

cwd()

Path | None GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.cwd

(

self

)

Definition at line 54 of file SubprocessBaseTest.py.

    def cwd(self) -> Path | None:
        cwd = self.popen_kwargs.get("cwd")
        return Path(cwd) if cwd else None
 

expand_vars_from()

str GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.expand_vars_from ( str value, dict[str, str] env )

static

Definition at line 97 of file SubprocessBaseTest.py.

    def expand_vars_from(value: str, env: dict[str, str]) -> str:
        return Template(value).safe_substitute(env)
 

resolve_path()

str GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.resolve_path	(		cls,
		Path | str	path )

Resolve the given path to an absolute path,
expanding environment variables.
If path looks relative and does not point to anything
it is not modified.

Definition at line 59 of file SubprocessBaseTest.py.

    def resolve_path(cls, path: Path | str) -> str:
        """
        Resolve the given path to an absolute path,
        expanding environment variables.
        If path looks relative and does not point to anything
        it is not modified.
        """
        if isinstance(path, Path):
            path = str(path)
        path = os.path.expandvars(path)
 
        # handle the special case "path/to/file:some_suffix"
        suffix = ""
        if ":" in path:
            path, suffix = path.rsplit(":", 1)
            suffix = f":{suffix}"
 
        if not os.path.isabs(path):
            base_dir = file_path_for_class(cls).parent
            possible_path = str((base_dir / path).resolve())
            if os.path.exists(possible_path):
                path = possible_path
        return path + suffix
 

run_program()

FixtureResult GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.run_program	(		cls,
			tmp_path = None )

Run the specified program and capture its output.

Definition at line 186 of file SubprocessBaseTest.py.

    def run_program(cls, tmp_path=None) -> FixtureResult:
        """
        Run the specified program and capture its output.
        """
        start_time = datetime.now()
        command, env = cls._prepare_execution(tmp_path=tmp_path)
 
        proc = subprocess.Popen(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env=env,
            **cls.popen_kwargs,
        )
 
        stdout_chunks, stderr_chunks = [], []
        stdout = stderr = ""
        exceeded_stream = stack_trace = run_exception = None
        streams = {
            proc.stdout.fileno(): (stdout_chunks, "stdout"),
            proc.stderr.fileno(): (stderr_chunks, "stderr"),
        }
 
        def read_output():
            nonlocal stdout, stderr, exceeded_stream
            while not exceeded_stream and proc.poll() is None:
                readable, _, _ = select.select(streams.keys(), [], [], cls.timeout)
                for fileno in readable:
                    data = os.read(fileno, 1024)
                    chunks, stream_name = streams[fileno]
                    chunks.append(data)
                    if sum(len(chunk) for chunk in chunks) > STDOUT_LIMIT:
                        exceeded_stream = stream_name
                        break
 
            stdout = b"".join(stdout_chunks)
            stderr = b"".join(stderr_chunks)
 
        thread = threading.Thread(target=read_output)
        thread.start()
        thread.join(cls.timeout)
 
        if thread.is_alive():
            stack_trace = cls._handle_timeout(proc)
            run_exception = ProcessTimeoutError("Process timed out", stack_trace)
        elif exceeded_stream:
            run_exception = ExceededStreamError(
                "Stream exceeded size limit", exceeded_stream
            )
 
        end_time = datetime.now()
 
        completed_process = subprocess.CompletedProcess(
            args=command,
            returncode=proc.returncode,
            stdout=stdout,
            stderr=stderr,
        )
 
        # Only record  environment variables that match at least one
        # of the regular expressions in env_to_record.
        # If env_to_record is empty or None, do not record any env variable
        env_to_record = list(cls.env_to_record) if cls.env_to_record else []
        # extend env_to_record with the regular expressions in environment variables
        # named GAUDI_ENV_TO_RECORD, GAUDI_ENV_TO_RECORD_1, GAUDI_ENV_TO_RECORD_2, ...
        env_to_record.extend(
            env[name] for name in env if re.match(r"GAUDI_ENV_TO_RECORD(_\d+)?", name)
        )
        env = {
            key: value
            for key, value in env.items()
            if any(re.match(exp, key) for exp in env_to_record)
        }
        return FixtureResult(
            completed_process=completed_process,
            start_time=start_time,
            end_time=end_time,
            run_exception=run_exception,
            command=cls.command,
            expanded_command=command,
            env=env,
            cwd=cls.popen_kwargs["cwd"],
        )
 

run_program_for_dbg()

GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.run_program_for_dbg	(		cls,
			tmp_path )

Definition at line 271 of file SubprocessBaseTest.py.

    def run_program_for_dbg(cls, tmp_path):
        tmp_path = cls.popen_kwargs.get("cwd", tmp_path)
        command, env = cls._prepare_execution(tmp_path=tmp_path)
        print("Running the command: ", command)
        subprocess.run(
            command,
            env=env,
            cwd=tmp_path,
            **cls.popen_kwargs,
        )
 

test_fixture_setup()

None GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.test_fixture_setup	(		self,
		Callable[[str, str], None]	record_property,
		FixtureResult	fixture_result,
		Path | None	reference_path )

Record properties and handle any failures during fixture setup.

Definition at line 283 of file SubprocessBaseTest.py.

    ) -> None:
        """
        Record properties and handle any failures during fixture setup.
        """
        for key, value in fixture_result.to_dict().items():
            if value is not None:
                record_property(key, value)
        if reference_path:
            record_property("reference_file", str(reference_path))
 
        if fixture_result.run_exception:
            pytest.fail(f"{fixture_result.run_exception}")
 

test_returncode()

None GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.test_returncode	(		self,
		int	returncode )

Test that the return code matches the expected value.

Definition at line 302 of file SubprocessBaseTest.py.

    def test_returncode(self, returncode: int) -> None:
        """
        Test that the return code matches the expected value.
        """
        assert returncode == self.returncode

unset_vars()

None GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.unset_vars ( dict[str, str] env, list[str] vars_to_unset )

static

Definition at line 101 of file SubprocessBaseTest.py.

    def unset_vars(env: dict[str, str], vars_to_unset: list[str]) -> None:
        for var in vars_to_unset:
            env.pop(var, None)
 

update_env()

None GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.update_env	(		cls,
		dict[str, str]	env )

Definition at line 84 of file SubprocessBaseTest.py.

    def update_env(cls, env: dict[str, str]) -> None:
        if cls.environment:
            for item in cls.environment:
                key, value = item.split("=", 1)
                env[key] = cls.expand_vars_from(value, env)
 

Member Data Documentation

command

list GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.command = None

static

Definition at line 42 of file SubprocessBaseTest.py.

env_to_record

list GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.env_to_record

static

Initial value:

=  [
        r"^(.*PATH|BINARY_TAG|CMTCONFIG|PWD|TMPDIR|DISPLAY|LC_[A-Z]+|LANGUAGE|.*ROOT|.*OPTS|GAUDI_ENV_TO_RECORD(_\d+)?)$"
    ]

Definition at line 49 of file SubprocessBaseTest.py.

environment

list GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.environment = None

static

Definition at line 44 of file SubprocessBaseTest.py.

popen_kwargs

GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.popen_kwargs = {}

static

Definition at line 47 of file SubprocessBaseTest.py.

reference

str GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.reference = None

static

Definition at line 43 of file SubprocessBaseTest.py.

returncode

int GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.returncode = 0

static

Definition at line 46 of file SubprocessBaseTest.py.

timeout

GaudiTesting.SubprocessBaseTest.SubprocessBaseTest.timeout = 600

static

Definition at line 45 of file SubprocessBaseTest.py.

---

The documentation for this class was generated from the following file:

• GaudiPolicy/python/GaudiTesting/SubprocessBaseTest.py