auxiliary

Auxiliary common Interface Definition

module:: auxiliary
synopsis:: base auxiliary interface

class pykiso.auxiliary.AuxiliaryCommon[source]

Class use to encapsulate all common methods/attributes for both multiprocessing and thread auxiliary interface.

Auxiliary common attributes initialization.

abort_command(blocking=True, timeout_in_s=25)[source]

Force test to abort.

Parameters:

blocking (bool) – If you want the command request to be blocking or not
timeout_in_s (float) – Number of time (in s) you want to wait for an answer

Return type:

bool

Returns:

True - Abort was a success / False - if not

create_copy(*args, **config)[source]

Create a copy of the actual auxiliary instance with the new desired configuration.

Note

only named arguments have to be used

Warning

the call of create_copy will automatically suspend the current auxiliary until the it copy is destroyed

Parameters:: config (dict) – new desired auxiliary configuration
Return type:: AuxiliaryCommon
Returns:: a brand new auxiliary instance
Raises:: Exception – if positional parameters is given or unknown named parameters are given

abstract create_instance()[source]

Handle auxiliary creation.

Return type:: bool

abstract delete_instance()[source]

Handle auxiliary deletion.

Return type:: bool

destroy_copy()[source]: Stop the current auxiliary copy and resume the original. :rtype: None

Warning

stop the copy auxiliary will automatically start the base/original one

lock_it(timeout_in_s)[source]

Lock to ensure exclusivity.

Parameters:: timeout_in_s (float) – How many second you want to wait for the lock
Return type:: bool
Returns:: True - Lock done / False - Lock failed

resume()[source]: Resume current auxiliary’s run, by running the create_instance method in the background. :rtype: None

Warning

due to the usage of create_instance if an issue occurred the exception AuxiliaryCreationError is raised.

abstract run()[source]

Run function of the auxiliary.

Return type:: None

run_command(cmd_message, cmd_data=None, blocking=True, timeout_in_s=0)[source]

Send a test request.

Parameters:

cmd_message (Union[Message, bytes, str]) – command request to the auxiliary
cmd_data (Any) – data you would like to populate the command with
blocking (bool) – If you want the command request to be blocking or not
timeout_in_s (int) – Number of time (in s) you want to wait for an answer

Return type:

bool

Returns:

True - Successfully sent / False - Failed by sending / None

stop()[source]

Force the thread to stop itself.

Return type:: None

suspend()[source]

Supend current auxiliary’s run.

Return type:: None

unlock_it()[source]

Unlock exclusivity

Return type:: None

wait_and_get_report(blocking=False, timeout_in_s=0)[source]

Wait for the report of the previous sent test request.

Parameters:

blocking (bool) – True: wait for timeout to expire, False: return immediately
timeout_in_s (int) – if blocking, wait the defined time in seconds

Return type:

Union[Message, bytes, str]

Returns:

a message.Message() - Message received / None - nothing received