1.1.3.1.2.1.1. test_automation.targets.fvp.autofvpnetworking

Module to manage telnet sessions

1.1.3.1.2.1.1.1. Attributes

logger

1.1.3.1.2.1.1.2. Classes

TelnetSessionManager

Manage multiple telnet sessions for exposed telnet ports of FVP.

1.1.3.1.2.1.1.3. Module Contents

test_automation.targets.fvp.autofvpnetworking.logger[source]

class test_automation.targets.fvp.autofvpnetworking.TelnetSessionManager(config=None, platform=None, **options)[source]

Manage multiple telnet sessions for exposed telnet ports of FVP.

Read FVP boot log to get terminal-port mappings
Spawn pexpect based telnet session for each terminal
Continuously log for each telnet session’s output
Check for login and shell prompts, send commands and redirect outputs.
Drain logs and clean up sessions on demand.

Parameters:: platform (Optional[str])

test_name: str | None[source]

_init_load_platform_cfg(config, platform)[source]

Load the platform configuration from the provided configuration object.

Parameters:

config – Configuration object exposing get_platform().
platform (str) – Platform key to retrieve from the configuration.

Returns:

Platform configuration dictionary returned by get_platform().

Raises:

KeyError – If the platform key is not found.

Return type:

dict

_init_load_timeouts(platform_cfg, platform)[source]

Initialize login and shell prompt timeouts from platform configuration.

Parameters:

platform_cfg (dict) – Platform configuration dictionary containing timeout settings.
platform (str) – Platform name (used for error messages).

Raises:

KeyError – If required timeout keys are missing.

Return type:

None

_init_load_prompts(platform_cfg, platform)[source]

Initialize default and per-terminal prompt strings from configuration.

Parameters:

platform_cfg (dict) – Platform configuration dictionary containing prompt mappings.
platform (str) – Platform name (used for error messages).

Raises:

KeyError – If default prompt keys are missing.

Return type:

None

_init_required_terminals(platform_cfg, platform)[source]

Initialize the list of required terminal names.

Parameters:

platform_cfg (dict) – Platform configuration dictionary.
platform (str) – Platform name (used for error messages).

Raises:

KeyError – If required_terminals is missing or empty.

Return type:

None

_init_port_map(platform_cfg)[source]

Initialize terminal-to-port mapping for the current platform.

Parameters:: platform_cfg (dict) – Platform configuration dictionary containing port mappings.
Return type:: None

_init_paths_and_state(base_log_root, timeout, encoding)[source]

Initialize filesystem paths, log directories, and runtime state.

Parameters:

base_log_root (str) – Base directory path for all telnet log files.
timeout (int) – Default timeout in seconds for telnet interactions.
encoding (str) – Encoding used for log and I/O operations.

Return type:

None

_get_log_path(port, terminal_name=None)[source]

Get the log file path for a telnet session.

The log filename is constructed as <terminal_name>_<port>.log when a terminal name is provided. If no terminal name is available, it falls back to telnet_<port>.log.

Parameters:

port (int) – Telnet port number
terminal_name (Optional[str]) – Optional terminal name

Returns:

Full path to the log file

Return type:

str

create_and_start_session(port, terminal_name=None)[source]

Spawn a telnet session and start logging simultaneously.

Parameters:

port (int) – Telnet port of the telnet session.
terminal_name (Optional[str]) – Optional name for prompt mapping.

Return type:

None

get_port(terminal_name)[source]

Return the port number for a terminal name, or None if not registered.

Parameters:: terminal_name (str) – Terminal name to query.
Returns:: Port number if known, else None.
Return type:: Optional[int]

pause_logging(port)[source]

Pause the background logging thread for a port.

This allows direct expect() calls on the session without the logging thread consuming the output.

Parameters:: port (int) – Port number to pause logging for.
Return type:: None

resume_logging(port)[source]

Resume the background logging thread for a port.

Parameters:: port (int) – Port number to resume logging for.
Return type:: None

_log_telnet_session(port, session)[source]

Continuously log data from the session into its logfile, logs errors and exits cleanly when the session or logfile is closed.

Parameters:

port (int) – Port number of the telnet session.
session (pexpect.spawn) – Active pexpect telnet session instance.

Return type:

None

_log_iteration(port, session)[source]

One non-blocking expect iteration from the original logic.

Parameters:

port (int)
session (pexpect.spawn)

Return type:

bool

_get_terminal_name(port, terminal_name)[source]

Resolve the terminal name for a given port.

Returns the explicit terminal_name if provided. Otherwise, uses _term_for_port() to resolve the name.

Parameters:

port (int)
terminal_name (Optional[str])

Return type:

Optional[str]

wait_for_prompt_in_log(*args, **kwargs)[source]

Waits for a prompt/pattern in the telnet log file.

If regex is True: prompt is treated as a regex pattern.
If regex is False: prompt is treated as a literal string.
If regex is None: a heuristic decides if prompt looks like
a regex (auto-detect).

Parameters:

args – Positional arguments (port, prompt, timeout).
kwargs – Keyword arguments used to pass options (port=…, prompt=…, timeout=…).

Returns:

True if the prompt/pattern is found in the log, otherwise False.

Return type:

bool

_extract_wait_args(args, kwargs)[source]

Extract and validate wait_for_prompt_in_log arguments.

This helper pops known keyword arguments, applies positional args, validates there are no unknown kwargs, and ensures port and prompt are present.

Parameters:

args (tuple) – Positional args passed through.
kwargs (dict) – Keyword args passed through (will be mutated).

Returns:

Tuple (port, prompt, timeout, regex, ignore_case).

Raises:

TypeError – on invalid or missing arguments.

Return type:

tuple[int, str, int, bool | None, bool]

_pop_wait_kwargs(kwargs)[source]

Pop recognized keyword arguments for the wait helper.

Recognized keys: port, prompt, timeout, regex, ignore_case.

Parameters:: kwargs (dict) – Keyword arguments dict (mutated by popping).
Returns:: (port, prompt, timeout, regex, ignore_case).
Return type:: tuple[int | None, str | None, int, bool | None, bool]

_apply_positional_args(args, current)[source]

Apply positional arguments to (port, prompt, timeout).

Parameters:

args (tuple) – Positional args tuple.
current (tuple[int | None, str | None, int]) – Tuple (port, prompt, timeout).

Returns:

Updated (port, prompt, timeout).

Raises:

TypeError – If more than three positional args are provided.

Return type:

tuple[int | None, str | None, int]

_check_for_regex(text)[source]

Determine if a string looks like a regex.

Checks for common regex markers such as groups, classes, escapes, quantifiers and anchors.

Parameters:: text (str) – User given string.
Returns:: True if the string likely represents a regex pattern.
Return type:: bool

_compile_prompt_pattern(port, prompt, options)[source]

Compile the prompt into a regex.Pattern.

Parameters:

port (int) – Port number (used only for logging).
prompt (str) – Prompt text or regex pattern.
options (tuple[bool, bool]) – Tuple (regex, ignore_case).

Returns:

Compiled pattern, or None on error.

Return type:

re.Pattern | None

_poll_until_found(wait_params)[source]

Poll the log file until the pattern is found or timeout occurs.

Parameters:: wait_params (tuple[int, str, int, bool, str, re.Pattern]) – Tuple (port, prompt, timeout, regex, log_path, prompt_pattern).
Returns:: True if match found, otherwise False.
Return type:: bool

_is_timeout_reached(now, deadline)[source]

Return True if the current time has passed the deadline.

Parameters:

now (float) – Current time (seconds since epoch).
deadline (float) – Deadline timestamp.

Returns:

True if timeout reached, otherwise False.

Return type:

bool

_update_debug_status_if_needed(debug_params)[source]

Update debug status if enough time has passed since last debug.

Parameters:: debug_params (tuple[int, float, float, int, float, float]) – Tuple (port, now, started_at, timeout, last_debug_time, debug_interval).
Returns:: Updated last debug time.
Return type:: float

_file_contains(path, regex)[source]

Return matched text if pattern is found, else False.

Parameters:

path (str) – Path to the log file to be scanned.
regex (re.Pattern) – Compiled regular expression pattern to search for.

Returns:

The matched text if the pattern is found, else False.

_debug_wait_status(port, started_at, timeout)[source]

Log a periodic debug message while waiting for a prompt.

Parameters:

port (int) – Telnet port number associated with the session.
started_at (float) – Timestamp when waiting started.
timeout (int) – Total timeout value in seconds.

Returns:

None

Return type:

None

login_if_needed(port, login_prompt=None, shell_prompt=None)[source]

Ensure a shell is ready on the given port, logging in if required.

Parameters:

port (int) – Telnet port number.
login_prompt (Optional[str]) – Optional login prompt regex/string override.
shell_prompt (Optional[str]) – Optional shell prompt regex/string override.

Returns:

True if a shell prompt is observed; otherwise False.

Return type:

bool

_resolve_prompts(term_name, login_prompt, shell_prompt)[source]

Resolve effective login and shell prompts for a terminal.

Parameters:

term_name (Optional[str]) – Logical terminal name, if known.
login_prompt (Optional[str]) – Optional login prompt override.
shell_prompt (Optional[str]) – Optional shell prompt override.

Returns:

Tuple where the first item may be None if no login is expected.

Return type:

Tuple[Optional[str], str]

_handle_login_path(port, login_prompt, shell_prompt)[source]

Handle the login flow if a login prompt is expected and observed.

Parameters:

port (int) – Telnet port number.
login_prompt (Optional[str]) – Login prompt regex/string, if any.
shell_prompt (str) – Expected shell prompt regex/string.

Returns:

True if shell prompt is reached; else False.

Return type:

bool

_await_shell_after_login(port, shell_prompt)[source]

Wait for the shell prompt after credentials were submitted.

Parameters:

port (int) – Telnet port number.
shell_prompt (str) – Expected shell prompt regex/string.

Returns:

True if the shell prompt appears; else False.

Return type:

bool

_handle_no_login_path(port, shell_prompt)[source]

Ensure the shell is present when a login prompt is not expected/found.

Parameters:

port (int) – Telnet port number.
shell_prompt (str) – Expected shell prompt regex/string.

Returns:

True if the shell prompt is found; else False.

Return type:

bool

_term_for_port(port)[source]

Resolve the logical terminal name for a given telnet port.

Parameters:: port (int) – Telnet port number.
Returns:: The terminal name mapped to port or None if unknown.
Return type:: Optional[str]

execute_command_with_prompt_capture(port, command, timeout=None)[source]

Execute a shell command on the specified Telnet session, wait for its completion marker, and capture both exit code and output.

Parameters:

port (int) – Telnet port number associated with the active session.
command (str) – The shell command to be executed.
timeout (Optional[int]) – Optional timeout in seconds for command execution. If not provided, the default session timeout is used.

Returns:

A tuple (exit_code, output) where: - exit_code is the integer exit status of the executed command. - output is the cleaned textual output of the command.

Raises:

RuntimeError – If the Telnet session is not alive or terminal mapping cannot be resolved.

Return type:

Tuple[int, str]

_get_session_and_term(port)[source]

Fetch the live pexpect session and its logical terminal name.

Parameters:: port (int) – Telnet port number whose session is required.
Returns:: Tuple of (session, terminal_name).
Raises:: RuntimeError – If the session is not alive or the terminal name cannot be resolved.
Return type:: Tuple[pexpect.spawn, str]

_get_prompts_for_term(term_name)[source]

Resolve the effective login and shell prompts for a terminal.

Parameters:: term_name (str) – Logical terminal name.
Returns:: Tuple (login_prompt, shell_prompt); the first may be None when no login is expected.
Raises:: RuntimeError – If no shell prompt can be determined.
Return type:: Tuple[Optional[str], str]

_ensure_shell_ready(port, prompts, timeout)[source]

Ensure a shell is ready on the session, logging in if needed.

Parameters:

port (int) – Telnet port number.
prompts (Tuple[Optional[str], str]) – Tuple (login_prompt, shell_prompt).
timeout (int) – Seconds to wait for prompts.

Returns:

True if a shell prompt is confirmed; else False.

Return type:

bool

_create_full_command(command)[source]

Build the final command and its completion marker.

Parameters:: command (str) – Shell command to execute.
Returns:: Tuple (full_cmd, marker_literal, marker_regex) where marker_literal is the plain prefix (e.g. "CMD_DONE_") and marker_regex captures the exit code.
Return type:: Tuple[str, str, re.Pattern]

_send_and_try_echo(port, full_cmd, marker_literal)[source]

Send the full command and wait for the marker echo.

Parameters:

port (int) – Telnet port number.
full_cmd (str) – Composite shell command to send.
marker_literal (str) – Plain marker prefix expected in the echo.

Returns:

None.

Return type:

None

_await_completion(port, expect, timeout)[source]

Wait for the command execution to complete by detecting either the completion marker or the shell prompt in the session output.

Parameters:

port (int) – Telnet port number of the session.
expect (Tuple[str, re.Pattern]) – Tuple (shell_prompt, marker_regex).
timeout (int) – Maximum number of seconds to wait for completion.

Returns:

A tuple (exit_code, error_buffer).

Return type:

Tuple[Optional[int], Optional[str]]

_expect_marker_or_prompt(port, expect, timeout)[source]

Waits for either the marker or the shell prompt.

:returns(“marker”, session) when marker matched (use session.match): (“prompt”, buffer) when shell prompt matched (session.before) (“timeout”, buffer) on timeout (session.before)

Parameters:

port (int)
expect (Tuple[str, re.Pattern])
timeout (int)

Return type:

Tuple[str, object]

_parse_exit_from_match(session, marker_regex)[source]

Extract the command exit code from the session match object.

Parameters:

session (pexpect.spawn) – The pexpect session whose last match contains the exit code.
marker_regex (re.Pattern) – Compiled regex pattern that captures the exit code from the completion marker.

Returns:

A tuple (exit_code, None). Falls back to scanning the session buffer if the marker match cannot be parsed.

Return type:

Tuple[int, None]

_parse_exit_from_buffer(buffer, marker_regex, port)[source]

Parse the exit code from a plain buffer when the shell prompt appears.

Parameters:

buffer (str) – The text preceding the shell prompt.
marker_regex (re.Pattern) – Compiled regex pattern used to locate and extract the exit code.
port (int) – Telnet port number of the session (used for logging).

Returns:

A tuple (exit_code, error_buffer): - exit_code is the parsed integer exit status, or None if the marker was missing or invalid. - error_buffer is the raw buffer when no marker is found, otherwise None.

Return type:

Tuple[Optional[int], Optional[str]]

_finalize_output_and_persist(raw_output, meta)[source]

Clean command output, remove any buffer, and copy the result to file.

Parameters:

raw_output (str) – Raw session output captured before the prompt or marker.
meta (Tuple[str, re.Pattern, int, str]) – Tuple (full_cmd, marker_regex, port, command).

Returns:

The cleaned command output string.

Return type:

str

_ensure_shell_or_login(port, prompts, timeout)[source]

Ensure a usable shell, logging in first if necessary.

Parameters:

port (int) – Telnet port of the session.
prompts (Tuple[Optional[str], str]) – Tuple (login_prompt, shell_prompt).
timeout (int) – Seconds to wait for shell prompt if needed.

Returns:

True if shell is ready, else False.

Return type:

bool

_expect_pattern(session, pattern, timeout)[source]

Wait for a pattern, returning whether it matched and any partial buffer.

Parameters:

session (pexpect.spawn) – Active pexpect session.
pattern (str) – Regex/string to expect.
timeout (int) – Seconds to wait for a match.

Returns:

Tuple (matched, buffer_on_timeout) where matched is True if the pattern matched; otherwise False and buffer_on_timeout contains session.before.

Return type:

Tuple[bool, str]

_already_at_shell(port, shell_prompt)[source]

Check quickly whether we are already at the shell prompt.

Parameters:

port (int) – Telnet port of the session.
shell_prompt (str) – Expected shell prompt regex/string.

Returns:

True if the shell prompt is present, else False.

Return type:

bool

_login_path_then_shell(port, prompts, timeout)[source]

Attempt the login path, then wait for the shell prompt.

Parameters:

port (int) – Telnet port of the session.
prompts (Tuple[str, str]) – Tuple (login_prompt, shell_prompt).
timeout (int) – Seconds to wait for the shell prompt if needed.

Returns:

True if shell is ready, else False.

Return type:

bool

_after_saw_login_send_root_and_wait_shell(port, shell_prompt)[source]

Send credentials after seeing the login prompt and wait for the shell.

Parameters:

port (int) – Telnet port of the session.
shell_prompt (str) – Expected shell prompt regex/string.

Returns:

True if shell prompt appears, else False.

Return type:

bool

_no_login_prompt_then_wait_shell(port, shell_prompt, timeout)[source]

If login prompt is not observed, wait directly for the shell prompt.

Parameters:

port (int) – Telnet port of the session.
shell_prompt (str) – Expected shell prompt regex/string.
timeout (int) – Seconds to wait for the shell prompt.

Returns:

True if shell prompt appears, else False.

Return type:

bool

_wait_for_shell_with_timeout(port, shell_prompt, timeout)[source]

Wait for a shell prompt within a time span.

Parameters:

port (int) – Telnet port of the session.
shell_prompt (str) – Expected shell prompt regex/string.
timeout (int) – Seconds to wait for the shell prompt.

Returns:

True if shell prompt appears, else False.

Return type:

bool

_fallback_exit_from_buffer(buffer, marker_regex, default=1)[source]

Try to extract the exit code from a buffer using the marker regex.

Parameters:

buffer (str) – Text captured before the prompt/marker.
marker_regex (re.Pattern) – Compiled regex that captures the exit code in group 1 (e.g. CMD_+DONE_(\d+)).
default (int) – Exit code to return when no code can be parsed.

Returns:

Parsed integer exit code, or default if unavailable.

Return type:

int

_clean_exec_output(raw_output, full_cmd, marker_regex)[source]

Remove noise from raw command output and return a clean string.

Parameters:

raw_output (str) – Unfiltered text collected before the marker/prompt.
full_cmd (str) – Exact command string that was sent (for echo removal).
marker_regex – Compiled regex for the completion marker line.

Returns:

Cleaned command output with noise removed.

Return type:

str

_detect_test_name()[source]

Decide a test name for logging, preferring an explicit name.

Returns:: Resolved test name or "<no test>".
Return type:: str

_preferred_test_name()[source]

Return the preferred explicit test name when available.

Returns:: Test name if set, else None.
Return type:: Optional[str]

_first_test_func_from_stack()[source]

Scan the call stack and return the first test_* function name.

Returns:: Name of the first function starting with test_, or None if no such frame exists.
Return type:: Optional[str]

_persist_cmd_output(port, command, output)[source]

Append a command’s output to the persistent command log file.

Parameters:

port (int) – Telnet port from which the command was executed.
command (str) – Exact shell command that was run.
output (str) – Final cleaned output to be written.

Returns:

None

Return type:

None

_check_prompt(term, port)[source]

Attempt to log in on required port, then wait for its shell prompt.

Parameters:

term (str) – Terminal name.
port (int) – Port number.

Return type:

None

_normalize_terminal(raw)[source]

Normalize a terminal key into the canonical terminal_* name.

Parameters:: raw (str) – Raw terminal key from the FVP driver.
Returns:: Lowercased name prefixed with terminal_ if missing.
Return type:: str

start_telnet_sessions_after_fvp_ready(fvp_driver, log_file_path='')[source]

With LocalFVP: poll get_ports() briefly so multiple terminals can appear, then spawn Telnet sessions for required terminals; if none match, fall back to all.

Parameters:

fvp_driver – Controller exposing terminal ports.
log_file_path (str) – Path to FVP boot log.

Return type:

None

_resolve_log_path(fvp_driver, log_file_path)[source]

Resolve the boot log path using the driver if possible.

Parameters:

fvp_driver – FVP driver which may implement log_path().
log_file_path (str) – Fallback path to use if the driver raises or does not implement the method.

Returns:

Resolved log path or "<unknown>" if none is available.

Return type:

str

_discover_ports(fvp_driver)[source]

Poll the driver for terminal→port mappings until they stabilize.

The poll runs up to min(timeout, 15) seconds, returning the most recent normalized mapping. It stops early if all required terminals are present.

Parameters:: fvp_driver – Driver exposing get_ports() -> Dict[str, int].
Returns:: Dict mapping normalized terminal names to ports.

_spawn_required_then_fallback(norm)[source]

Spawn Telnet sessions for required terminals, or all as a fallback.

Parameters:: norm (Dict[str, int]) – Mapping of normalized terminal names to ports.
Returns:: Mapping of terminal names to ports that actually started.
Return type:: Dict[str, int]

_wait_for_started_prompts(started)[source]

Wait for prompts on each started terminal. Spawns a thread per started terminal that checks the prompt, then joins all threads before returning.

Parameters:: started (Dict[str, int]) – Mapping of terminal name to telnet port that has been started.
Returns:: None
Return type:: None

close_sessions()[source]

Gracefully close all active telnet sessions and their log files.

Returns:: None
Return type:: None

set_log_dir(path)[source]

Set the base directory for telnet logs and initialize output file.

Creates the directory if needed and ensures the command output file exists. Raises a RuntimeError on permission or OS errors.

Parameters:: path (str) – New base directory for logs.
Returns:: None
Raises:: RuntimeError – If the directory or file cannot be created.
Return type:: None

drain_console_to_tail(session, timeout)[source]

Consume any pending console output to start from a clean state.

param session: The pexpect session to read from. param timeout: Maximum time to spend draining the console.

Parameters:: session (pexpect.spawn)
Return type:: None

run_cmd(port, cmd, timeout=None)[source]

Execute a command on the target console and return its output.

Port:

Telnet port number of the target console.

Parameters:

cmd (str) – Command string to execute.
timeout (Optional[int]) – Maximum time to wait for command completion.
port (int)

Returns:

Command output with leading and trailing whitespace removed.

Raises:

RuntimeError – If the command returns a non-zero status code.

Return type:

str