7. record_auxiliary

Example can be found here Passively record a channel.

7.1. 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)

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) – deprecated, will not be taken into account.

• manual_start_record (bool) – flag to not start recording on auxiliary creation

clear_buffer()

Clean the buffer that contain received messages.

Return type:

None

dump_to_file(filename, mode='w+', data=None)

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 (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()

Return the entire log buffer content.

Return type:

str

Returns:

buffer content

is_log_empty()

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)

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)

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()

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)

Decode the received bytes

Parameters:

data (bytes) – data to be decoded

Return type:

str

Returns:

data decoded

previous_log()

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()

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)

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)

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)

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)

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()

Clear buffer and start recording.

Return type:

None

stop_recording()

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)

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)

Constructor

get_data()

Get data from the string

Return type:

str

Returns:

data from the string

set_data(data)

Add data to the already existing data string

Parameters:

data (str) – the data to be write over the existing string

Return type:

None