1.1.3.1.1.1.1. test_automation.targets.fpga.autofpganetworking

Auto FPGA networking manager.

This module provides an SSH based session manager used to boot and interact with an FPGA target.

1.1.3.1.1.1.1.1. Attributes

logger
RUN_DIR_RE
_UART_LOG_NAME_RE
_PROMPT_RE
_LOGIN_MARKER

1.1.3.1.1.1.1.2. Classes

SSHConfig	SSH connection settings for the FPGA host.
_OutputTarget	Destination for execution output.
_OutputBuffer	Accumulates execution output text between recv() calls
AutoFPGANetworking	SSH based networking manager for FPGA targets.

1.1.3.1.1.1.1.3. Functions

_read_all(stdout)

Read and decode all data from a Paramiko stdout-like stream.

1.1.3.1.1.1.1.4. Module Contents

test_automation.targets.fpga.autofpganetworking.logger

test_automation.targets.fpga.autofpganetworking.RUN_DIR_RE

test_automation.targets.fpga.autofpganetworking._UART_LOG_NAME_RE

test_automation.targets.fpga.autofpganetworking._PROMPT_RE

test_automation.targets.fpga.autofpganetworking._LOGIN_MARKER = 'login:'

test_automation.targets.fpga.autofpganetworking._read_all(stdout)

Read and decode all data from a Paramiko stdout-like stream.

Parameters:

stdout – Stream object returned by Paramiko exec_command.

Returns:

Decoded text (UTF-8 with replacement on decode errors).

Return type:

str

class test_automation.targets.fpga.autofpganetworking.SSHConfig

SSH connection settings for the FPGA host.

Parameters:

• host – Remote hostname or IP.

• username – SSH username.

• password – Optional password used for password or interactive auth.

• port – SSH port.

• connect_timeout_s – Connection/auth timeout in seconds.

host: str

username: str

password: str | None = None

port: int = 22

connect_timeout_s: int = 20

class test_automation.targets.fpga.autofpganetworking._OutputTarget

Destination for execution output.

Parameters:

• log_file – local boot log file handle

• on_line – optional callback invoked per output record

log_file: any

on_line: Callable[[str], None] | None = None

class test_automation.targets.fpga.autofpganetworking._OutputBuffer

Accumulates execution output text between recv() calls until complete log records can be flushed.

text: str = ''

class test_automation.targets.fpga.autofpganetworking.AutoFPGANetworking(cfg, *args, **kwargs)

Bases: test_automation.utils.networking_base.BaseNetworkManager

SSH based networking manager for FPGA targets.

Provides remote session management, command execution, log streaming, and console interaction utilities required to control and monitor an FPGA platform during automated testing.

Parameters:

cfg (SSHConfig)

_resolve_platform_name(args, kwargs)

Resolve platform_name from kwargs or first positional arg.

Parameters:

• args – Positional args passed to the constructor.

• kwargs – Keyword args passed to the constructor.

Returns:

Platform name as string.

Return type:

str

_resolve_log_paths(kwargs)

Resolve LogPathOptions from kwargs (log_paths) or overrides legacy.

Parameters:

kwargs – Keyword args passed to the constructor.

Returns:

Structured log path configuration.

Return type:

test_automation.targets.fpga.fpga_runtime_options.LogPathOptions

cfg

platform_name = ''

_override_log_dir

_override_log_prefix = ''

_client: paramiko.SSHClient | None = None

_connected = False

_last_run_dir: str | None = None

_run_dir_lock

_run_dir_event

_run_ts: str | None = None

_run_local_dir: pathlib.Path | None = None

_remote_run_logs_local_dir: pathlib.Path | None = None

_boot_log_local: pathlib.Path | None = None

_run_thread: threading.Thread | None = None

_run_exit_status: int | None = None

_run_chan: paramiko.Channel | None = None

_initialize_ssh_client()

Create and configure a Paramiko SSH client.

Returns:

Configured SSH client instance.

Return type:

paramiko.SSHClient

_authenticate_with_password(client)

Attempt SSH authentication using password-based authentication.

Parameters:

client (paramiko.SSHClient) – Initialized SSH client.

Raises:

• paramiko.AuthenticationException – If password auth fails.

• Exception – For other SSH transport/connection errors.

Return type:

None

_handle_interactive_prompts(_title, _instructions, prompt_list)

Paramiko keyboard-interactive callback.

For prompts containing the word “password” (case-insensitive), the configured password is returned; otherwise an empty response is used.

Parameters:

• _title – Prompt title (unused).

• _instructions – Prompt instructions (unused).

• prompt_list – List of (prompt, echo) tuples.

Returns:

List of response strings matching prompt_list length

_authenticate_with_interactive(client)

Attempt SSH authentication using keyboard-interactive authentication.

Parameters:

client (paramiko.SSHClient) – Initialized SSH client.

Raises:

RuntimeError – If the SSH transport cannot be obtained.

Return type:

None

connect()

Establish an SSH connection to the FPGA host.

The method first attempts standard password authentication and falls back to keyboard based authentication when required by the server.

Raises:

RuntimeError – If SSH connectivity or authentication fails.

Return type:

None

disconnect()

Close SSH connection and clear internal state.

This closes the Paramiko client (if present) and resets the connected flag.

Return type:

None

_ensure_ssh_client()

Ensure the SSH client is connected.

Raises:

RuntimeError – If the SSH client is not connected.

Return type:

None

_upload_binary(src, remote_archive)

Upload an archive to the remote host using SFTP.

Parameters:

• src (pathlib.Path) – Local archive path.

• remote_archive (str) – Destination path on the remote host.

Raises:

RuntimeError – If SSH is not connected.

Return type:

None

_build_extract_cmd(src_name, remote_base_dir)

Build an extraction command for a supported archive type.

Parameters:

• src_name (str) – Archive file name (not full path).

• remote_base_dir (str) – Remote directory to extract into.

Returns:

Shell command string used for extraction.

Raises:

RuntimeError – If the archive extension is unsupported.

Return type:

str

_detect_fpga_workdir(remote_base_dir)

Detect a top-level directory created by extracting the archive and lists directories under remote_base_dir and uses the most recently modified entry as a candidate work directory. If no directory is found, the base directory is returned.

Parameters:

remote_base_dir (str) – Remote extraction base directory.

Returns:

Detected work directory path.

Raises:

RuntimeError – If SSH is not connected.

Return type:

str

copy_payloads_to_remote(local_archive, remote_base_dir)

Upload an archive to the remote host, extract it, and return workdir.

Parameters:

• local_archive (str) – Local path to an archive containing FPGA binary.

• remote_base_dir (str) – Remote directory where the archive will be uploaded and extracted.

Returns:

The top-level extracted directory if one is detected, otherwise remote_base_dir.

Raises:

• RuntimeError – If SSH is not connected.

• FileNotFoundError – If local_archive does not exist.

• RuntimeError – If extraction fails.

Return type:

str

_ensure_local_run_dirs()

Ensure local directories exist for boot logs and downloaded RUN_DIR logs.

Creates: - <CWD>/logs/<platform>_<ts>/ - <CWD>/logs/<platform>_<ts>/remote_run_logs/ - <CWD>/logs/<platform>_<ts>/<platform>_boot_<ts>.log

Return type:

None

property boot_log_path: pathlib.Path | None

Return the local boot log file path.

Returns:

Path to the local boot log file, if initialized.

Return type:

Optional[pathlib.Path]

property local_run_dir: pathlib.Path | None

Return the local run directory created for this execution.

Returns:

Path to the local run directory, if initialized.

Return type:

Optional[pathlib.Path]

property local_remote_run_logs_dir: pathlib.Path | None

Return the local directory where remote RUN_DIR logs are downloaded.

Returns:

Path to the local remote logs directory, if initialized.

Return type:

Optional[pathlib.Path]

property last_run_dir: str | None

Return the most recently detected remote RUN_DIR.

Returns:

Remote RUN_DIR string, if detected.

Return type:

Optional[str]

_create_full_cmd(env_setup_cmds, remote_cmd, remote_workdir)

Compose the full remote command string.

Parameters:

• env_setup_cmds (Sequence[str]) – Sequence of shell commands to prepare the environment (e.g. source ...).

• remote_cmd (str) – Main command to execute on the remote host.

• remote_workdir (Optional[str]) – Optional remote working directory to cd into before executing remote_cmd.

Returns:

Fully composed command string suitable for Paramiko exec.

Return type:

str

_open_exec_channel(full_cmd)

Open a PTY-enabled Paramiko session channel and execute full_cmd.

Parameters:

full_cmd (str) – Command string returned by _create_full_cmd().

Returns:

Paramiko channel used to stream output and obtain exit value.

Raises:

RuntimeError – If SSH is not connected or transport is missing.

Return type:

paramiko.Channel

_write_and_process_line(text, log_file, on_line)

Write output text to the boot log, invoke callback, and detect RUN_DIR.

Parameters:

• text (str) – Text chunk/record to write.

• log_file – Open file handle to write to.

• on_line (Optional[Callable[[str], None]]) – Optional callback invoked with the written text.

Return type:

None

_flush_complete_records(sink, buffer)

Flush complete newline-terminated records from the output buffer.

Parameters:

• sink (_OutputTarget) – Output sink configuration.

• buffer (_OutputBuffer) – Output buffer to flush.

Return type:

None

_process_output_log(chunk, sink, buffer)

Decode a received byte chunk and append it to the output buffer.

Parameters:

• chunk (bytes) – Bytes read from Paramiko channel.

• sink (_OutputTarget) – Output sink configuration.

• buffer (_OutputBuffer) – Output buffer to append to.

Return type:

None

_finalize_run_output(exec_stream, sink, buffer)

Drain remaining output after process termination and record exit status.

Parameters:

• exec_stream (paramiko.Channel) – Paramiko channel.

• sink (_OutputTarget) – Output sink configuration.

• buffer (_OutputBuffer) – Output buffer.

Return type:

None

_read_available_output_once(exec_handle, sink, buffer)

Read and process available output once (non-blocking).

Parameters:

• exec_handle (paramiko.Channel) – Paramiko channel.

• sink (_OutputTarget) – Output sink configuration.

• buffer (_OutputBuffer) – Output buffer.

Return type:

None

_monitor_execution_output(*, exec_handle, on_line)

Monitor output from a Paramiko channel and stream to boot log.

This method runs until the remote command exits and writes all received output into the local boot log file.

Parameters:

• exec_handle (paramiko.Channel) – Paramiko channel that is executing remote command.

• on_line (Optional[Callable[[str], None]]) – Optional callback invoked for each flushed log record.

Return type:

None

start_remote_run(run_spec=None, on_line=None, **kwargs)

Start the remote run command in a background thread and stream output to the local boot log.

Parameters:

• run_spec (Optional[test_automation.targets.fpga.fpga_runtime_options.RemoteRunSpec]) – Optional structured run specification. If not provided, legacy keyword arguments are used to construct it.

• on_line (Optional[Callable[[str], None]]) – Optional callback invoked for each flushed output.

• kwargs – Backward-compatible keyword arguments.

Raises:

RuntimeError – If connect() was not called first.

Return type:

None

wait_for_run_dir(timeout_s=180)

Wait until RUN_DIR is detected from the remote output stream.

Parameters:

timeout_s (int) – Maximum time to wait in seconds.

Returns:

Detected RUN_DIR path.

Raises:

TimeoutError – If RUN_DIR is not detected within timeout.

Return type:

str

wait_remote_exit(timeout_s=None)

Wait for the remote run thread to complete and return exit status.

Parameters:

timeout_s (Optional[int]) – Optional timeout for joining the run thread.

Returns:

Exit status if available, otherwise None.

Return type:

Optional[int]

find_uart_login(run_dir, *, tail_lines=120)

Search UART logs under the given RUN_DIR for a login prompt.

This method scans all UART_log_vuart_* files located under run_dir and inspects the last tail_lines lines of each log for the presence of a login marker (e.g., "login:"). If a match is found, the corresponding VUART index, PTS number, and log path are returned.

Parameters:

• run_dir (str) – Remote RUN_DIR path containing UART log files.

• tail_lines (int) – Number of lines from the end of each UART log to inspect. Defaults to 120.

Returns:

Tuple (vuart, pts, uart_log_path) if a login prompt is detected; otherwise None.

Return type:

Optional[Tuple[int, int, str]]

Raises:

RuntimeError – If the SSH client is not connected.

send_to_vuart_link_sftp(run_dir, payload, **kwargs)

Write payload to RUN_DIR/vuart_<vuart>__dev_pts_<pts> using SFTP. This helper writes directly to the VUART link to avoid shell quoting issues.

Parameters:

• run_dir (str) – Remote RUN_DIR path.

• payload (str) – Text to write to the VUART link.

• kwargs – Backward-compatible keyword args.

Raises:

• TypeError – If vuart or pts are missing.

• RuntimeError – If SSH is not connected.

Return type:

None

wait_for_shell_prompt(uart_log_path, options=None, **kwargs)

Wait until a root shell prompt is visible in the specified UART log.

This method periodically tails the UART log file and checks for a shell prompt pattern. It stops when the prompt is detected or when the timeout expires.

Parameters:

• uart_log_path (str) – Remote UART log file path.

• options (Optional[test_automation.targets.fpga.fpga_runtime_options.ShellPromptWaitOptions]) – Optional structured wait configuration. If not provided, values may be supplied via backward-compatible keyword arguments.

• timeout_s – (kwarg) Maximum time in seconds to wait for the shell prompt. Defaults to 120.

• poll_s – (kwarg) Polling interval in seconds between log checks. Defaults to 1.0.

• tail_lines – (kwarg) Number of log lines to inspect on each poll. Defaults to 200.

Returns:

None

Raises:

• RuntimeError – If the SSH client is not connected.

• TimeoutError – If the shell prompt is not detected within the timeout.

Return type:

None

wait_login_and_send_root(options=None, **kwargs)

Wait for a UART login prompt, then wait for a shell prompt.

This method scans available UART logs for a login marker, sends the root username to the detected VUART link, and then waits until a shell prompt becomes available.

Parameters:

• options (Optional[test_automation.targets.fpga.fpga_runtime_options.LoginWaitOptions]) – Optional structured login wait configuration. If not provided, values may be supplied via backward-compatible keyword arguments.

• timeout_s – (kwarg) Maximum time in seconds to wait for the login prompt. Defaults to 900.

• poll_s – (kwarg) Polling interval in seconds between login checks. Defaults to 2.0.

• tail_lines – (kwarg) Number of UART log lines to inspect during each poll. Defaults to 150.

• shell_prompt_timeout_s – (kwarg) Maximum time in seconds to wait for the shell prompt after sending root. Defaults to 120.

Returns:

Tuple (uart_log_path, vuart, pts) identifying the UART log file and its associated VUART and PTS numbers.

Returns:

Tuple[str, int, int]

Raises:

• RuntimeError – If the SSH client is not connected.

• TimeoutError – If the login prompt or shell prompt is not detected within the timeout.

Return type:

Tuple[str, int, int]

run_uart_command(cmd, **kwargs)

Send a command to a specific UART by writing to its VUART link.

The command is appended with a newline and written to the corresponding RUN_DIR/vuart_<vuart>__dev_pts_<pts> link.

Parameters:

• cmd (str) – Command string to send to the UART.

• vuart – (kwarg) VUART index.

• pts – (kwarg) PTS number associated with the VUART.

Returns:

None

Raises:

RuntimeError – If the SSH client is not connected.

Return type:

None

list_uart_logs()

List UART log files under the current RUN_DIR.

Returns:

Sorted list of tuples (vuart, pts, uart_log_path).

Raises:

RuntimeError – If SSH is not connected or RUN_DIR is unknown.

Return type:

List[Tuple[int, int, str]]

read_uart_log(uart_log_path)

Read the full contents of a UART log file.

Parameters:

uart_log_path (str) – Remote UART log file path.

Returns:

UART log contents (UTF-8 decoded with replacement).

Raises:

RuntimeError – If SSH is not connected.

Return type:

str

tail_uart_log(uart_log_path, *, lines=200)

Read the tail of a UART log file.

Parameters:

• uart_log_path (str) – Remote UART log file path.

• lines (int) – Number of lines from the end of the file to return.

Returns:

UART log tail (UTF-8 decoded with replacement).

Raises:

RuntimeError – If SSH is not connected.

Return type:

str

find_uarts_with_strings(strings, *, tail_lines=300)

Find UART logs whose tail contains any of the provided strings.

Parameters:

• strings (List[str]) – List of substrings to search for.

• tail_lines (int) – Number of tail lines to inspect for each UART log.

Returns:

List of tuples (vuart, pts, uart_log_path) that matched.

Raises:

RuntimeError – If SSH is not connected.

Return type:

List[Tuple[int, int, str]]

start_uart_keepalive(**kwargs)

This can help keep UART sessions active and stimulate console output while waiting for prompts.

Return type:

None

stop_uart_keepalive()

Stop the UART keepalive thread if it is running.

Return type:

None

map_uarts_by_predicate(predicate, *, tail_lines=300)

Apply a predicate to UART log text and return matching UARTs.

Parameters:

• predicate (Callable[[str], bool]) – Callable predicate(text) -> bool.

• tail_lines (int) – Number of lines from the UART log tail to inspect.

Returns:

List of tuples (vuart, pts, uart_log_path).

Return type:

List[Tuple[int, int, str]]

download_run_dir_logs()

Download all files under the current remote RUN_DIR to local logs.

The destination directory is: <CWD>/logs/<platform>_<ts>/remote_run_logs/

Returns:

Local destination path for downloaded remote logs.

Raises:

RuntimeError – If SSH is not connected or RUN_DIR is unknown.

Return type:

pathlib.Path

_sftp_get_dir_recursive(sftp, remote_dir, local_dir)

Recursively download a remote directory using SFTP.

Parameters:

• sftp (paramiko.SFTPClient) – Active Paramiko SFTP client.

• remote_dir (str) – Remote directory path to download.

• local_dir (pathlib.Path) – Local directory to write downloaded files into.

Return type:

None

static _shell_escape(s)

Escape a string for safe inclusion in a single-quoted shell literal.

Parameters:

s (str) – Input string.

Returns:

Single-quoted, shell-safe literal.

Return type:

str