Method execute_script

src/test/core/process_manager.py:428–707

Execute a single script Args: script_config: script configuration run_time: run time (seconds), if None use script configuration Returns: ProcessInfo: process information

(self, script_config: ScriptConfig, run_time: Optional[int] = None)

        return conn

    def execute_script(self, script_config: ScriptConfig, run_time: Optional[int] = None) -> ProcessInfo:
        """
        Execute a single script

        Args:
            script_config: script configuration
            run_time: run time (seconds), if None use script configuration

        Returns:
            ProcessInfo: process information
        """
        script_run_time = run_time or script_config.time_sec

        # Prepare execution environment
        env = os.environ.copy()
        env.update(script_config.environment)

        # Determine working directory
        if script_config.cwd:
            cwd = self.base_cwd / script_config.cwd
        else:
            cwd = self.base_cwd

        # Build command
        cmd = [script_config.path] + script_config.args

        print(f"🚀 Execute script: {script_config.path}")
        print(f"    Working directory: {cwd}")
        print(f"    Run time: {script_run_time} seconds")
        print(f"    Environment variables: {script_config.environment}")

        if not self._is_remote(script_config):
            try:
                master_fd, slave_fd = pty.openpty()
                process = subprocess.Popen(
                    cmd,
                    cwd=str(cwd),
                    env=env,
                    stdin=slave_fd,
                    stdout=slave_fd,
                    stderr=slave_fd,
                    close_fds=True
                )
                try:
                    os.close(slave_fd)
                except Exception:
                    pass
                process_info = ProcessInfo(
                    script_path=script_config.path,
                    pid=process.pid,
                    process=process,
                    start_time=datetime.now(),
                    backend="local",
                    pty_master_fd=master_fd
                )

                with self._lock:
                    self._processes[script_config.path] = process_info