8. record_auxiliary

Example can be found here Passively record a channel.

8.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)[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