pypts API documentation¶

This section provides detailed documentation for the public modules and classes within the pypts framework.

Core Runner (pts)¶

class pypts.pts.PtsApi(input_queue: queue.Queue, event_queue: _queue.SimpleQueue, recipe_queue: _queue.SimpleQueue, sequence_name: str | None = None, on_start: object = None, on_stop: object = None)¶

Bases: object

event_queue: SimpleQueue¶

input_queue: Queue¶

on_start: object = None¶

on_stop: object = None¶

Starts a PTS recipe running in a separate thread and returns an API object to interact with it. Also starts a background thread (report_listener) to generate a CSV report (report.csv) incrementally in the ~/pts_reports/ directory.

Parameters:

recipe_file (str) – Path to the YAML recipe file to run
sequence_name (str, optional) – Name of the sequence in the recipe to run. Defaults to “Main”.

Returns:

An object containing queues for interacting with the running recipe:

input_queue: Queue for sending commands to the recipe
event_queue: Queue for receiving events from the recipe
recipe_queue: Queue for receiving recipe execution reports
on_start: callable set by the GUI layer; invoked when a START command arrives
on_stop: callable set by the GUI layer; invoked when a STOP/EXIT command arrives

Return type:

PtsApi

recipe_queue: SimpleQueue¶

sequence_name: str | None = None¶

pypts.pts.run_pts(sequence_name: str | None = None) → PtsApi¶

Recipe Components (recipe)¶

class pypts.recipe.Recipe(recipe_file_path, event_sender=None)¶

Bases: object

Represents and executes a test recipe defined in a multi-document YAML file.

Loads the recipe structure, manages global variables, sequences, and overall execution flow. The detailed structure of the recipe YAML file is described in Recipe YAML Format.

Recipe files must validate as recipe language 2.0.0 before any executable runtime state is constructed.

classmethod from_definition(definition: Recipe, source_name: str = '<definition>', event_sender=None)¶: Construct executable state from an already validated aggregate model.

run(runtime: Runtime, sequence_name: str | None = None)¶

Executes the main sequence of the recipe.

Sets up the runtime, determines the serial number, runs the specified sequence, sends pre/post recipe events, sends the STOP_LISTENER signal to the report queue, and returns the collected results.

Parameters:

runtime (Runtime) – The runtime environment.
sequence_name (str, optional) – The name of the sequence to start execution from. Defaults to “Main”.
serial_number (str, optional) – An explicit serial number to use. If None, prompts the user.
get_serial_number_func (callable, optional) – A custom function to get the serial number.

Returns:

A list of the top-level StepResult objects generated during the run.

Return type:

List[StepResult]

property total_steps: int¶: Count the result rows this recipe is expected to produce.

class pypts.recipe.ResultType(*values)¶

Bases: IntEnum

DONE = 1¶

ERROR = 4¶

FAIL = 3¶

PASS = 2¶

SKIP = 0¶

STOP = 5¶

class pypts.recipe.Runtime(event_queue, report_queue)¶

Bases: object

append_result(parent_step_id: UUID, result: StepResult)¶

continue_on_error: bool¶

current_sequence_name: str¶

get_global(index)¶

get_globals()¶

get_local(name)¶

get_results()¶

get_sequence(name)¶

pop_locals()¶

push_locals(locals)¶

pypts_version: str¶

recipe_continue_on_error: bool | None¶

recipe_file_name: str¶

recipe_name: str¶

results: List[StepResult]¶

send_event(event_name: str, *event_data)¶

serial_number: str¶

set_global(name, value)¶

set_globals(globals)¶

set_local(name, value)¶

set_sequences(sequences)¶

stop_event = <threading.Event at 0x7f4a14c1c690: unset>¶

test_package: str¶

class pypts.recipe.Sequence(definition: Sequence)¶

Bases: object

run(runtime: Runtime, input: dict, parent_step: UUID = None)¶

class pypts.recipe.Step(step_name, id='', description='', input_mapping=None, output_mapping=None, skip=False, critical=False, continue_on_error=False)¶

Bases: object

static build_step(step_definition: CommonStepDefinition)¶

Build an executable step from a validated recipe step definition.

This is the single boundary between declarative recipe data and executable runtime behavior. The definition has already passed the Pydantic and aggregate semantic validation performed by the parser.

Parameters:: step_definition – Validated Pydantic step definition.
Returns:: Fully configured executable Step.

check_indexing()¶

handle_step_abort(step_result, runtime, input, reason='Stopped by user')¶

input_mapping: dict¶

is_critical()¶

is_skipped()¶

output_mapping: dict¶

process_inputs(runtime: Runtime)¶

process_outputs(runtime: Runtime, step_output: dict)¶

run(runtime: Runtime, input, parent_step: UUID = None, stop_event=None)¶

Executes the step, handling setup, execution, error handling, and output processing.

Processes inputs, calls the internal _step method, processes outputs, handles potential errors, creates a StepResult, sends pre/post events, and sends the StepResult to the report_queue.

Parameters:

runtime (Runtime) – The current execution runtime environment.
input – The input data for the step (not used directly here, processed in process_inputs).
parent_step (uuid.UUID, optional) – The UUID of the parent step, if any.

Returns:

An object containing the results of the step execution.

Return type:

StepResult

static run_steps(runtime: Runtime, step_list: List[Self], parent_step: UUID, stop_event=None) → List[StepResult]¶

class pypts.recipe.StepResult(step=None, parent=None)¶

Bases: object

append_subresult(subresult: Self)¶

error_info: str¶

static evaluate_multiple_step_results(step_results: List[Self]) → ResultType¶

get_result()¶

static get_result_by_uuid(step_results: List[Self], uuid: UUID) → Self¶

image_paths: List[str]¶

inputs: dict¶

is_type(result_type: ResultType)¶

outputs: dict¶

parent: UUID¶

print_result(indent='')¶

pypts_version: str¶

recipe_file_name: str¶

recipe_name: str¶

result: ResultType¶

sequence_name: str¶

serial_number: str¶

set_error(error_info=None, inputs={})¶

set_result(result_type=ResultType.DONE, inputs={}, outputs={})¶

set_skip()¶

set_stop(error_info=None, inputs={})¶

step: Step¶

subresults: List[StepResult]¶

uuid: UUID¶

Step Implementations (steps)¶

class pypts.steps.IndexedStep(step: Step, **kwargs)¶

Bases: Step

A step that wraps another step (template_step) and runs it multiple times based on indexed inputs. It aggregates results and outputs.

check_indexing()¶: IndexedStep itself doesn’t have indexed inputs; it manages the execution based on the template step’s inputs. This method shouldn’t be relevant after instantiation.

class pypts.steps.PythonModuleStep(action_type: str, module: str, method_name: str = None, continue_on_error: bool = False, **kwargs)¶

Bases: Step

Executes a method or interacts with attributes within a specified Python module.

class pypts.steps.SSHCloseStep(continue_on_error=False, **kwargs)¶

Bases: Step

Closes an existing Paramiko SSH connection.

class pypts.steps.SSHConnectStep(continue_on_error=False, **kwargs)¶

Bases: Step

Attempts an SSH connection to the given host and returns connection status.

class pypts.steps.SSHUploadStep(files: list, permissions=493, skip_if_sha256_match: bool = True, local_package: str = None, continue_on_error: bool = False, **kwargs)¶

Bases: Step

Upload one or more local files to a remote host via SFTP.

Uses the ssh_client global populated by SSHConnectStep. Supports SHA-256 skip-if-match so repeated recipe runs do not re-upload files whose content has not changed — important for large binaries.

YAML fields:

files (list, required): List of {local: <path>, remote: <path>} mappings. local may be relative to the project root or, when local_package is set, relative to that installed package.
local_package (str, optional): Resolve local paths via importlib.resources.files(<package>) rather than the project root. Use this when the files live inside an installed Python package (e.g. packaged ARM binaries in a bin/ directory).
permissions (int or str, optional): chmod value applied after each upload. Accepts an integer (e.g. 493) or an octal string (e.g. "0o755"). Default: 0o755.
skip_if_sha256_match (bool, optional): Skip the upload when the local and remote SHA-256 digests are identical. Default: true.
continue_on_error (bool, optional): When true, an upload failure does not abort the recipe. Default: false.

Returns a dict {"passed": bool, "deployed": [names], "skipped": [names]}.

Example YAML:

- steptype: SSHUploadStep
  step_name: Deploy C tool binaries
  local_package: pts_diot_mfe
  files:
    - local: bin/mfe-acq
      remote: /tmp/mfe-acq
    - local: bin/adcreg
      remote: /tmp/adcreg
  permissions: "0o755"
  skip_if_sha256_match: true
  continue_on_error: false
  output_mapping:
    passed:
      type: passfail

class pypts.steps.SequenceStep(sequence: dict, **kwargs)¶

Bases: Step

Executes another sequence (defined either internally in the same recipe or potentially externally) as a single step within the current sequence.

class pypts.steps.SerialNumberStep(continue_on_error: bool = False, **kwargs)¶

Bases: Step

Prompts the user for a serial number via the UI and stores it in the runtime.

Sends a ‘get_serial_number’ event with a response queue. The UI handler must put the serial number string into that queue. The serial number is stored in both runtime.serial_number and the global variable serial_number.

class pypts.steps.UserInteractionStep(module: str = None, action_type: str = None, method_name: str = None, continue_on_error: bool = False, **kwargs)¶

Bases: Step

Pauses recipe execution and sends an event to request interaction from a user via a UI or other external interface. Waits for a response.

class pypts.steps.WaitStep(**kwargs)¶

Bases: Step

Pauses recipe execution for a specified duration.

Reporting (report)¶

Handles the generation of incremental CSV reports for pypts recipe execution.

class pypts.report.Report(output_dir: str | Path, timestamp, overwrite=True, image_dir_name: str = 'img')¶

Bases: object

Manages incremental CSV report generation during PTS execution. Writes results to a CSV file in the specified directory as they become available via the add_step_result method.

add_step_result(result: StepResult)¶: Adds a StepResult and writes it to the CSV report file.

finish_reports()¶: Closes the CSV report file handle.

pypts.report.generate_html_report(csv_path: Path, html_path: Path, output_dir: Path = None)¶: Reads a CSV report and generates an HTML version.

pypts.report.report_listener(result_queue: SimpleQueue, output_dir: str, overwrite: bool, include_serial_in_name: bool = False)¶

Listens to a queue for StepResult objects and generates reports incrementally.

Exits gracefully when the STOP_LISTENER sentinel object is received on the queue.

Sequence selection¶

Recipe.run(runtime, sequence_name=None) and run_pts(sequence_name=None) use the recipe’s configured main_sequence when no name is supplied. Passing an explicit sequence name overrides that default and unknown names raise a descriptive error. run_recipe_app(recipe_path, sequence_name=None) preloads the supplied recipe in the GUI and forwards the same sequence selection.

pypts API documentation¶

Core Runner (pts)¶

Recipe Components (recipe)¶

Step Implementations (steps)¶

Reporting (report)¶

Sequence selection¶

pypts

Navigation

Related Topics