Included Auxiliaries

pykiso comes with some ready to use implementations of different auxiliaries.

Collection of ready to use auxiliaries

module: auxiliaries

CommunicationAuxiliary

module: communication_auxiliary
synopsis: Auxiliary used to send raw bytes via a connector instead of pykiso.Messages

class pykiso.lib.auxiliaries.communication_auxiliary.CommunicationAuxiliary(com, **kwargs)[source]

Auxiliary used to send raw bytes via a connector instead of pykiso.Messages.

Constructor.

Parameters: com (CChannel) – CChannel that supports raw communication

receive_message(blocking=True, timeout_in_s=None)[source]

Receive a raw message.

Parameters

blocking (bool) – wait for message till timeout elapses?
timeout_in_s (Optional[float]) – maximum time in second to wait for a response

Return type

bytes

Returns

raw message

send_message(raw_msg)[source]

Send a raw message (bytes) via the communication channel.

Parameters: raw_msg (bytes) – message to send
Return type: bool
Returns: True if command was executed otherwise False

Example Auxiliary

module: example_test_auxiliary
synopsis: Example auxiliary that simulates a normal test run without ever doing anything.

Warning

Still under test

class pykiso.lib.auxiliaries.example_test_auxiliary.ExampleAuxiliary(name=None, com=None, flash=None, **kwargs)[source]

Example of an auxiliary implementation.

Constructor.

Parameters

name – Alias of the auxiliary instance
com – Communication connector
flash – flash connector

Virtual DUT simulation package

module: simulated_auxiliary
synopsis: provide a simple interface to simulate a device under test

This auxiliary can be used as a simulated version of a device under test.

The intention is to set up a pair of CChannels like a pipe, for example a CCUdpServer and a CCUdp bound to the same address. One side of this pipe is then connected to this virtual auxiliary, the other one to a real auxiliary.

The SimulatedAuxiliary will then receive messages from the real auxiliary just like a proper TestApp on a DUT would and answer them according to a predefined playbook.

Each predefined playbooks are linked with real auxiliary received messages, using test case and test suite ids (see simulation). A so called playbook, is a basic list of different Message instances where the content is adapted to the current context under test (simulate a communication lost, a test case run failure…). (see scenario). In order to increase playbook configuration flexibility, predefined and reusable responses are located into response_templates.

`pykiso.lib.auxiliaries.simulated_auxiliary.simulated_auxiliary`	Simulated Auxiliary
`pykiso.lib.auxiliaries.simulated_auxiliary.simulation`	Simulation
`pykiso.lib.auxiliaries.simulated_auxiliary.scenario`	Scenario
`pykiso.lib.auxiliaries.simulated_auxiliary.response_templates`	ResponseTemplates

Simulated Auxiliary

module: simulated_auxiliary
synopsis: auxiliary used to simulate a virtual Device Under Test(DUT)

class pykiso.lib.auxiliaries.simulated_auxiliary.simulated_auxiliary.SimulatedAuxiliary(com=None, **kwargs)[source]

Custom auxiliary use to simulate a virtual DUT.

Initialize attributes.

Parameters: com – configured channel

Simulation

module: simulation
synopsis: map virtual DUT behavior with test case/suite id

Warning

Still under test

class pykiso.lib.auxiliaries.simulated_auxiliary.simulation.Simulation[source]

Simulate a virtual DUT, by playing pre-defined scenario depending on test case and test suite id.

Initialize attributes and mapping.

get_scenario(test_suite_id, test_case_id)[source]

Return the selected scenario mapped with the received test case and test suite id.

Parameters

test_suite_id (int) – current test suite id
test_case_id (int) – current test case id

Return type

Scenario

Returns

scenario instance containing all steps

handle_default_response()[source]

Return a scenario to handle DUT default behavior.

Return type: Scenario
Returns: scenario instance containing all steps

handle_ping_pong()[source]

Return a scenario to handle init ping pong exchange.

Return type: Scenario
Returns: scenario instance containing all steps

Scenario

module: scenario
synopsis: base object used to create pre-defined virtual DUT scenario.

Warning

Still under test

class pykiso.lib.auxiliaries.simulated_auxiliary.scenario.Scenario(initlist=None)[source]: Container used to create pre-defined virtual DUT scenario.

class pykiso.lib.auxiliaries.simulated_auxiliary.scenario.TestScenario[source]

Encapsulate all possible test’s scenarios.

class VirtualTestCase[source]

Used to gather all virtual DUT test case scenarios based on their fixture level (setup, run, teardown).

class Run[source]

Used to gather all possible scenarios linked to a test case run execution.

classmethod handle_failed_report_run()[source]

Return a scenario to handle a complete test case with failed report at run phase.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_failed_report_run_with_log()[source]

Return a scenario to handle a complete test case with failed log and report at run phase.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_lost_communication_during_run_ack()[source]

Return a scenario to handle a complete test case with lost of communication during ACK to run Command.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_lost_communication_during_run_report()[source]

Return a scenario to handle a complete test case with lost of communication during report to run Command.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_not_implemented_report_run()[source]

Return a scenario to handle a complete test case with not implemented report at run phase.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_successful_report_run_with_log()[source]

Return a scenario to handle a complete test case with successful log and report at run phase.

Return type: Scenario
Returns: Scenario instance containing all steps

class Setup[source]

Used to gather all possible scenarios linked to a test case setup execution.

classmethod handle_failed_report_setup()[source]

Return a scenario to handle a complete test case with failed report at setup phase.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_lost_communication_during_setup_ack()[source]

Return a scenario to handle a complete test case with lost of communication during ACK to setup Command.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_lost_communication_during_setup_report()[source]

Return a scenario to handle a complete test case with lost of communication during report to setup Command.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_not_implemented_report_setup()[source]

Return a scenario to handle a complete test case with not implemented report at setup phase.

Return type: Scenario
Returns: Scenario instance containing all steps

class Teardown[source]

Used to gather all possible scenarios linked to a test case teardown execution.

classmethod handle_failed_report_teardown()[source]

Return a scenario to handle a complete test case with failed report at teardown phase.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_lost_communication_during_teardown_ack()[source]

Return a scenario to handle a complete test case with lost of communication during ACK to teardown Command.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_lost_communication_during_teardown_report()[source]

Return a scenario to handle a complete test case with lost of communication during report to teardown Command.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_not_implemented_report_teardown()[source]

Return a scenario to handle a complete test case with not implemented report at teardown phase.

Return type: Scenario
Returns: Scenario instance containing all steps

class VirtualTestSuite[source]

Used to gather all virtual DUT test suite scenarios based on their fixture level (setup, teardown).

class Setup[source]

Used to gather all possible scenarios linked to a test suite setup execution.

classmethod handle_failed_report_setup()[source]

Return a scenario to handle a test suite setup with report failed.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_lost_communication_during_setup_ack()[source]

Return a scenario to handle a lost of communication during ACK to setup command.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_lost_communication_during_setup_report()[source]

Return a scenario to handle a lost of communication during report to setup command.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_not_implemented_report_setup()[source]

Return a scenario to handle a test suite setup with report not implemented.

Return type: Scenario
Returns: Scenario instance containing all steps

class Teardown[source]

Used to gather all possible scenarios linked to a test suite teardown execution.

classmethod handle_failed_report_teardown()[source]

Return a scenario to handle a test suite teardown with report failed.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_lost_communication_during_teardown_ack()[source]

Return a scenario to handle a lost of communication during ACK to teardown command.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_lost_communication_during_teardown_report()[source]

Return a scenario to handle a lost of communication during report to teardown command.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_not_implemented_report_teardown()[source]

Return a scenario to handle a test suite teardown with report not implemented.

Return type: Scenario
Returns: Scenario instance containing all steps

classmethod handle_successful()[source]

Return a scenario to handle a complete successful test case exchange(TEST CASE setup->run->teardown).

Return type: Scenario
Returns: Scenario instance containing all steps

ResponseTemplates

module: response_templates
synopsis: Used to create a set of predefined messages

Warning

Still under test

class pykiso.lib.auxiliaries.simulated_auxiliary.response_templates.ResponseTemplates[source]

Used to create a set of predefined messages (ACK, NACK, REPORT …).

classmethod ack(msg)[source]

Return an acknowledgment message.

Parameters: msg (Message) – current received message
Return type: List[Message]
Returns: list of Message

classmethod ack_with_logs_and_report_nok(msg)[source]

Return an acknowledge message and log messages and report message with verdict failed + tlv part with failure reason.

param msg

current received message

return

list of Message

Return type: List[Message]

classmethod ack_with_logs_and_report_ok(msg)[source]

Return an acknowledge message and log messages and report message with verdict pass.

param msg

current received message

return

list of Message

Return type: List[Message]

classmethod ack_with_report_nok(msg)[source]

Return an acknowledgment message and a report message with verdict failed + tlv part with failure reason.

Parameters: msg (Message) – current received message
Return type: List[Message]
Returns: list of Message

classmethod ack_with_report_not_implemented(msg)[source]

Return an acknowledge message and a report message with verdict test not implemented.

Parameters: msg (Message) – current received message
Return type: List[Message]
Returns: list of Message

classmethod ack_with_report_ok(msg)[source]

Return an acknowledgment message and a report message with verdict pass.

Parameters: msg (Message) – current received message
Return type: List[Message]
Returns: list of Message

classmethod default(msg)[source]

handle default response, if not test case/suite run just return ACK message otherwise ACK + REPORT.

Parameters: msg (Message) – current received message
Return type: Message
Returns: list of Message

classmethod get_random_reason()[source]

Return tlv dictionary containing a random reason from pre-defined reason list.

Parameters: msg – current received message
Return type: dict
Returns: tlv dictionary with failure reason

classmethod nack_with_reason(msg)[source]

Return a NACK message with a tlv part containing the failure reason.

Parameters: msg (Message) – current received message
Return type: List[Message]
Returns: list of Message

Instrument Control Auxiliary

module: instrument_control
synopsis: provide a simple interface to control instruments using SCPI protocol.

The functionalities provided in this package may be used directly inside ITF tests using the corresponding auxiliary, but also using a CLI.

Warning

This auxiliary can only be used with the cc_visa or cc_tcp_ip connector.
It is not intended to be used with a proxy connector.
One instrument is bound to one auxiliary even if the instrument has multiple channels.

`pykiso.lib.auxiliaries.instrument_control_auxiliary.instrument_control_auxiliary`	Instrument Control Auxiliary
`pykiso.lib.auxiliaries.instrument_control_auxiliary.instrument_control_cli`	Instrument Control CLI
`pykiso.lib.auxiliaries.instrument_control_auxiliary.lib_scpi_commands`	Library of SCPI commands
`pykiso.lib.auxiliaries.instrument_control_auxiliary.lib_instruments`	Library of instruments communicating via VISA

Instrument Control Auxiliary

module: instrument_control_auxiliary
synopsis: Auxiliary used to communicate via a VISA connector using the SCPI protocol.

class pykiso.lib.auxiliaries.instrument_control_auxiliary.instrument_control_auxiliary.InstrumentControlAuxiliary(com, instrument='', write_termination='\\n', output_channel=None, **kwargs)[source]

Auxiliary used to communicate via a VISA connector using the SCPI protocol.

Constructor.

Parameters

com (CChannel) – VISAChannel that supports VISA communication
instrument – name of the instrument currently in use (will be used to adapt the SCPI commands)
write_termination – write termination character
output_channel (Optional[int]) – output channel to use on the instrument currently in use (if more than one)

handle_query(query_command)[source]

Send a query request to the instrument. Uses the ‘query’ method of the: channel if available, uses ‘cc_send’ and ‘cc_receive’ otherwise.

Parameters: query_command (str) – query command to send
Return type: str
Returns: Response message, None if the request expired with a timeout.

handle_read()[source]

Handle read command by calling associated connector cc_receive.

Return type: str
Returns: received response from instrument otherwise empty string

handle_write(write_command, validation=None)[source]

Send a write request to the instrument and then returns if the value was successfully written. A query is sent immediately after the writing and the answer is compared to the expected one.

Parameters

write_command (str) – write command to send
validation (Optional[Tuple[str, Union[str, List[str]]]]) – tuple of the form (validation command (str), expected output (str or list of str))

Return type

str

Returns

status message depending on the command validation: SUCCESS, FAILURE or NO_VALIDATION

query(query_command)[source]

Send a query request to the instrument. Uses the ‘query’ method of the: channel if available, uses ‘cc_send’ and ‘cc_receive’ otherwise.

Parameters: query_command (str) – query command to send
Return type: Union[bytes, str]
Returns: Response message, None if the request expired with a timeout.

read()[source]

Send a read request to the instrument.

Return type: Union[str, bool]
Returns: received response from instrument otherwise empty string

write(write_command, validation=None)[source]

Send a write request to the instrument.

Parameters

write_command (str) – command to send
validation (Optional[Tuple[str, Union[str, List[str]]]]) – contain validation criteria apply on the response

Return type

str

Instrument Control CLI

module: instrument_control_cli
synopsis: Command Line Interface used to communicate with an instrument using the SCPI protocol.

class pykiso.lib.auxiliaries.instrument_control_auxiliary.instrument_control_cli.ExitCode(value)[source]: List of possible exit codes

class pykiso.lib.auxiliaries.instrument_control_auxiliary.instrument_control_cli.Interface(value)[source]: List of available interfaces

pykiso.lib.auxiliaries.instrument_control_auxiliary.instrument_control_cli.initialize_logging(log_level)[source]

Initialize the logging by setting the general log level

Parameters: log_level (str) – any of DEBUG, INFO, WARNING, ERROR
Return type: getLogger
Returns: configured Logger

pykiso.lib.auxiliaries.instrument_control_auxiliary.instrument_control_cli.parse_user_command(user_cmd)[source]

Parses the command from user input in interactive mode

Parameters: user_cmd (str) – command provided by the user in interactive mode
Return type: dict
Returns: a single-item dictionary containing the parsed command as key the the corresponding payload as value

pykiso.lib.auxiliaries.instrument_control_auxiliary.instrument_control_cli.perform_actions(instr_aux, actions)[source]

Performs the desired actions from the CLI arguments

Parameters

instr_aux (InstrumentControlAuxiliary) – instrument on which to perform the actions
actions (dict) – dictionary containing the parsed argument and the corresponding value.

Return type

None

pykiso.lib.auxiliaries.instrument_control_auxiliary.instrument_control_cli.setup_interface(interface, baud_rate=None, ip_address=None, port=None, protocol=None, name=None)[source]

Set up the instrument auxiliary with the appropriate interface settings. The ip address must be provided in the case of TCPIP interfaces, as must the serial port for VISA_SERIAL interface. The baud rate and the output channel to use are optional.

Parameters

interface (str) – interface to use
baud_rate (Optional[int]) – baud rate to use
ip_address (Optional[str]) – ip address of the remote instrument (used for remote control only)
port (Optional[int]) – the port of the device to connect to. This is either a serial port for a VISA_SERIAL interface or an IP port in case of an TCPIP interfaces.
protocol (Optional[str]) – The protocol to use for VISA_TCPIP interfaces.
name (Optional[str]) – instrument name used to adapt the SCPI commands to be sent to the instrument

Return type

InstrumentControlAuxiliary

Returns

The created instrument auxiliary.

Library of SCPI commands

module: lib_scpi_commands
synopsis: Library of helper functions used to send requests to instruments with SCPI protocol. This library can be used with any VISA instance having a write and a query method.

class pykiso.lib.auxiliaries.instrument_control_auxiliary.lib_scpi_commands.LibSCPI(visa_object, instrument='')[source]

Class containing common SCPI commands for write and query requests.

Constructor.

Parameters

visa_object – any visa object having a write and a query method
instrument (str) – name of the instrument in use. If registered, the commands adapted to this instrument’s capabilities are used instead of the default ones.

disable_output()[source]

Disable output on the currently selected output channel of an instrument.

Return type: str
Returns: the writing operation’s status code

enable_output()[source]

Enable output on the currently selected output channel of an instrument.

Return type: str
Returns: the writing operation’s status code

get_all_errors()[source]

Get all errors of an instrument.

return: list of off errors

get_command(cmd_tag, cmd_type, cmd_validation=None)[source]

Return the pre-defined command.

Parameters

cmd_tag (str) – command tag corresponding to the command to execute
cmd_type (str) – either ‘write’ or ‘query’
cmd_validation (Optional[tuple]) – expected output after validation (only used in write commands)

Return type

Tuple

Returns

the associated command plus a tuple containing the associated query and the expected response (if cmd_validation is not none) otherwise None

get_current_limit_high()[source]

Returns the current upper limit (in V) of an instrument.

Return type: str
Returns: the query’s response message

get_current_limit_low()[source]

Returns the current lower limit (in V) of an instrument.

Return type: str
Returns: the query’s response message

get_identification()[source]

Get the identification information of an instrument.

Returns: the instrument’s identification information

get_nominal_current()[source]

Query the nominal current of an instrument on the selected channel (in A).

Return type: str
Returns: the nominal current

get_nominal_power()[source]

Query the nominal power of an instrument on the selected channel (in W).

Return type: str
Returns: the nominal power

get_nominal_voltage()[source]

Query the nominal voltage of an instrument on the selected channel (in V).

Return type: str
Returns: the nominal voltage

get_output_channel()[source]

Get the currently selected output channel of an instrument.

Return type: str
Returns: the currently selected output channel

get_output_state()[source]

Get the output status (ON or OFF, enabled or disabled) of the currently selected channel of an instrument.

Return type: str
Returns: the output state (ON or OFF)

get_power_limit_high()[source]

Returns the power upper limit (in W) of an instrument.

Return type: str
Returns: the query’s response message

get_remote_control_state()[source]

Get the remote control mode (ON or OFF) of an instrument.

Returns: the remote control state

get_status_byte()[source]

Get the status byte of an instrument.

Returns: the instrument’s status byte

get_target_current()[source]

Get the desired output current (in A) of an instrument.

Return type: str
Returns: the target current

get_target_power()[source]

Get the desired output power (in W) of an instrument.

Return type: str
Returns: the target power

get_target_voltage()[source]

Get the desired output voltage (in V) of an instrument.

Return type: str
Returns: the target voltage

get_voltage_limit_high()[source]

Returns the voltage upper limit (in V) of an instrument.

Return type: str
Returns: the query’s response message

get_voltage_limit_low()[source]

Returns the voltage lower limit (in V) of an instrument.

Return type: str
Returns: the query’s response message

measure_current()[source]

Return the measured output current of an instrument (in A).

Return type: str
Returns: the measured current

measure_power()[source]

Return the measured output power of an instrument (in W).

Return type: str
Returns: the measured power

measure_voltage()[source]

Return the measured output voltage of an instrument (in V).

Return type: str
Returns: the measured voltage

reset()[source]

Reset an instrument.

Returns: NO_VALIDATION status code

self_test()[source]

Performs a self-test of an instrument.

Returns: the query’s response message

set_current_limit_high(limit_value)[source]

Set the current upper limit (in A) of an instrument.

Parameters: limit_value (float) – limit value to be set on the instrument
Return type: str
Returns: the writing operation’s status code

set_current_limit_low(limit_value)[source]

Set the current lower limit (in A) of an instrument.

Parameters: limit_value (float) – limit value to be set on the instrument
Return type: str
Returns: the writing operation’s status code

set_output_channel(channel)[source]

Set the output channel of an instrument.

Parameters: channel (int) – the output channel to select on the instrument
Return type: str
Returns: the writing operation’s status code

set_power_limit_high(limit_value)[source]

Set the power upper limit (in W) of an instrument.

Parameters: limit_value (float) – limit value to be set on the instrument
Return type: str
Returns: the writing operation’s status code

set_remote_control_off()[source]

Disable the remote control of an instrument. The instrument will respond to query and read commands only.

Returns: the writing operation’s status code

set_remote_control_on()[source]

Enables the remote control of an instrument. The instrument will respond to all SCPI commands.

Returns: the writing operation’s status code

set_target_current(value)[source]

Set the desired output current (in A) of an instrument.

Parameters: value (float) – value to be set on the instrument
Return type: str
Returns: the writing operation’s status code

set_target_power(value)[source]

Set the desired output power (in W) of an instrument.

Parameters: value (float) – value to be set on the instrument
Return type: str
Returns: the writing operation’s status code

set_target_voltage(value)[source]

Set the desired output voltage (in V) of an instrument.

Parameters: value (float) – value to be set on the instrument
Return type: str
Returns: the writing operation’s status code

set_voltage_limit_high(limit_value)[source]

Set the voltage upper limit (in V) of an instrument.

Parameters: limit_value (float) – limit value to be set on the instrument
Return type: str
Returns: the writing operation’s status code

set_voltage_limit_low(limit_value)[source]

Set the voltage lower limit (in V) of an instrument.

Parameters: limit_value (float) – limit value to be set on the instrument
Return type: str
Returns: the writing operation’s status code

Library of instruments communicating via VISA

module: lib_instruments
synopsis: Dictionaries containing the appropriate SCPI commands for some instruments.

Device Under Test Auxiliary

module: DUTAuxiliary
synopsis: The Device Under Test auxiliary allow to flash and run test on the target using the connector provided.

class pykiso.lib.auxiliaries.dut_auxiliary.DUTAuxiliary(name=None, com=None, flash=None, **kwargs)[source]

Device Under Test(DUT) auxiliary implementation.

Constructor.

Parameters

name (Optional[str]) – Alias of the auxiliary instance
com (Optional[CChannel]) – Communication connector
flash (Optional[Flasher]) – flash connector

resume()[source]

Resume DutAuxiliary’s run.

Set _is_suspend flag to False in order to re-activate flash sequence in case of e.g. a futur abort command.

Return type: None

suspend()[source]

Suspend DutAuxiliary’s run.

Set _is_suspend flag to True to avoid a re-flash sequence.

Return type: None

Proxy Auxiliary

module: proxy_auxiliary
synopsis: auxiliary use to connect multiple auxiliaries on a unique connector.

This auxiliary simply spread all commands and received messages to all connected auxiliaries. This auxiliary is only usable through proxy connector.

 ___________   ___________         ___________
|           | |           | ..... |           |
|   Aux 1   | |   Aux 1   |       |   Aux n   |
|___________| |___________|       |___________|
      |             |                   |
      |             |                   |
 ___________   ___________         ___________
|           | |           | ..... |           |
|Proxy_con 1| |Proxy_con 2|       |Proxy_con n|
|___________| |___________|       |___________|
      |             |                   |
      |             |                   |
      |             |                   |
 _____________________________________________
|                                             |
|               Proxy Auxiliary               |
|_____________________________________________|
                    |
                    |
 _____________________________________________
|                                             |
|               Connector Channel             |
|_____________________________________________|

class pykiso.lib.auxiliaries.proxy_auxiliary.ProxyAuxiliary(com, aux_list, activate_trace=False, trace_dir=None, trace_name=None, **kwargs)[source]

Proxy auxiliary for multi auxiliaries communication handling.

Initialize attributes.

Parameters

com (CChannel) – Communication connector
aux_list (List[str]) – list of auxiliary’s alias

get_proxy_con(aux_list)[source]

Retrieve all connector associated to all given existing Auxiliaries.

If auxiliary alias exists but auxiliary instance was not created yet, create it immediately using ConfigRegistry _aux_cache.

Parameters: aux_list (List[Union[str, AuxiliaryInterface]]) – list of auxiliary’s alias
Return type: Tuple[AuxiliaryInterface]
Returns: tuple containing all connectors associated to all given auxiliaries

run()[source]

Run function of the auxiliary thread

Return type: None

Multiprocessing Proxy Auxiliary

module: mp_proxy_auxiliary
synopsis: concrete implementation of a multiprocessing proxy auxiliary

This auxiliary simply spread all commands and received messages to all connected auxiliaries. This auxiliary is only usable through mp proxy connector.

 ___________   ___________         ___________
|           | |           | ..... |           |
|   Aux 1   | |   Aux 1   |       |   Aux n   |
|___________| |___________|       |___________|
      |             |                   |
      |             |                   |
 ___________   ___________         ___________
|           | |           | ..... |           |
|Proxy_con 1| |Proxy_con 2|       |Proxy_con n|
|___________| |___________|       |___________|
      |             |                   |
      |             |                   |
      |             |                   |
 _____________________________________________
|                                             |
|               Proxy Auxiliary               |
|_____________________________________________|
                    |
                    |
 _____________________________________________
|                                             |
|               Connector Channel             |
|_____________________________________________|

class pykiso.lib.auxiliaries.mp_proxy_auxiliary.MpProxyAuxiliary(com, aux_list, activate_trace=False, trace_dir=None, trace_name=None, **kwargs)[source]

Proxy auxiliary for multi auxiliaries communication handling.

..note :: this auxiliary version is using the multiprocessing: auxiliary interface.

Initialize attributes.

Parameters

com (CChannel) – Communication connector
aux_list (List[str]) – list of auxiliary’s alias
activate_trace (bool) – True if the trace is activate otherwise False
trace_dir (Optional[str]) – trace directory path (absolute or relative)
trace_name (Optional[str]) – trace full name (without file extension)

get_proxy_con(aux_list)[source]

Retrieve all connector associated to all given existing Auxiliaries.

If auxiliary alias exists but auxiliary instance was not created yet, create it immediately using ConfigRegistry _aux_cache.

Parameters: aux_list (List[str]) – list of auxiliary’s alias
Return type: Tuple[AuxiliaryInterface]
Returns: tuple containing all connectors associated to all given auxiliaries

run()[source]

Run function of the auxiliary process.

Return type: None

class pykiso.lib.auxiliaries.mp_proxy_auxiliary.TraceOptions(activate, dir, name)

Create new instance of TraceOptions(activate, dir, name)

property activate: Alias for field number 0

property dir: Alias for field number 1

property name: Alias for field number 2

Record Auxiliary

module: record_auxiliary
synopsis: Auxiliary used to record a connectors receive channel.

class pykiso.lib.auxiliaries.record_auxiliary.RecordAuxiliary(com, is_active=False, timeout=0, log_folder_path='', max_file_size=50000000, multiprocess=False, manual_start_record=False, **kwargs)[source]

Auxiliary used to record a connectors receive channel.

Constructor.

Parameters

com (CChannel) – Communication connector to record
is_active (bool) – Flag to actively poll receive channel in another thread
timeout (float) – timeout for the receive channel
log_path – path to the log folder
max_file_size (int) – maximal size of the data string
multiprocess (bool) –
use a Process instead of a Thread for active polling. Note1: the data will automatically be saved Note2: if proxy usage, all connectors should be ‘CCMpProxy’

and ‘processing’ flag set to True
manual_start_record (bool) – flag to not start recording on auxiliary creation

clear_buffer()[source]

Clean the buffer that contain received messages.

Return type: None

dump_to_file(filename, mode='w+', data=None)[source]

Writing data in file.

Parameters

filename (str) – name of the file where data are saved
mode (str) – modes of opening a file (eg: w, a)
data (Optional[str]) – Optional write/append specific data to the file.

Return type

bool

Returns

True if the dumping has been successful, False else

Raises

FileNotFoundError – if the given folder path is not a folder

get_data()[source]

Return the entire log buffer content.

Return type: str
Returns: buffer content

is_log_empty()[source]

Check if logs are available in the log buffer.

Return type: bool
Returns: True if log is empty, False either

is_message_in_full_log(message)[source]

Check for a message being in log.

Parameters: message (str) – message to check presence in logs.
Returns: True if a message is in log, False otherwise

is_message_in_log(message, from_cursor=True, set_cursor=True, display_log=False)[source]

Check for a message being in log.

Parameters

message (str) – str message to check presence in logs.
from_cursor (bool) – whether to get the logs from the last cursor position (True) or the full logs
set_cursor (bool) – whether to update the cursor
display_log (bool) – whether to log (via logging) the retrieved part or just return it

Return type

bool

Returns

True if a message is in log, False otherwise.

new_log()[source]

Get new entries (after cursor position) from the log. This will set the cursor.

Return type: str
Returns: return log after cursor

static parse_bytes(data)[source]

Decode the received bytes

Parameters: data (bytes) – data to be decoded
Return type: str
Returns: data decoded

previous_log()[source]

set cursor position to current position.

This will also display the logs from the last cursor position in the log.

Return type: str
Returns: log from the last current position

receive()[source]

Open channel and actively poll the connectors receive channel. Stop and close connector when stop receive event has been set.

Return type: None

search_regex_current_string(regex)[source]

Returns all occurrences found by the regex in the logs and: message received.

Parameters: regex (str) – str regex to compare to logs
Return type: Optional[List[str]]
Returns: list of matches with regular expression in the current string

search_regex_in_file(regex, filename)[source]

Returns all occurrences found by the regex in the logs and message received.

Parameters

regex (str) – str regex to compare to logs
filename (str) – filename of the desired file

Return type

Optional[List[str]]

Returns

list of matches with regular expression in the chosen file

search_regex_in_folder(regex)[source]

Returns all occurrences found by the regex in the logs and message received.

Parameters: regex (str) – str regex to compare to logs
Return type: Optional[Dict[str, List[str]]]
Returns: dictionary with filename and the list of matches with regular expression
Raises: FileNotFoundError – if the given folder path is not a folder

set_data(data)[source]

Add data to the already existing data string.

Parameters: data (str) – the data to be write over the existing string
Return type: None

start_recording()[source]

Clear buffer and start recording.

Return type: None

stop_recording()[source]

Stop recording.

Return type: None

wait_for_message_in_log(message, timeout=10.0, interval=0.1, from_cursor=True, set_cursor=True, display_log=False, exception_on_failure=True)[source]

Poll log at every interval time, fail if messages has not shown up within the specified timeout and exception set to True, log an error otherwise.

Parameters

message (str) – str message expected to show up
timeout (float) – int timeout in seconds for the check
interval (float) – int period in seconds for the log poll
from_cursor (bool) – whether to get the logs from the last cursor position (True) or the full logs
set_cursor (bool) – whether to update the cursor to the last log position
display_log (bool) – whether to log (via logging) the retrieved part or just return it
exception_on_failure (bool) – if set, raise a TimeoutError if the expected messages wasn’t found in the logs. Otherwise, simply output a warning.

Return type

bool

Returns

True if the message have been received in the log, False otherwise

Raises

TimeoutError – when a given message has not arrived in time

class pykiso.lib.auxiliaries.record_auxiliary.StringIOHandler(multiprocess=False)[source]

Constructor

Parameters: multiprocess (bool) – use a thread or multiprocessing lock.

get_data()[source]

Get data from the string

Return type: str
Returns: data from the string

set_data(data)[source]

Add data to the already existing data string

Parameters: data (str) – the data to be write over the existing string
Return type: None