utils.functions module

utils.functions.check_date_in_localized_interval(start_timestamp: int | None, end_timestamp: int | None, check_timestamp: int, check_timezone: str, interval_timezone: str) → bool

Checks if a timestamp is within an interval localizing with timezones

Parameters:

  • start_timestamp – bottom limit of the interval, if None no bottom limit is set

  • end_timestamp – top limit of the interval, if None no top limit is set

  • check_timestamp – timestamp checked in the interval

  • check_timezone – timezone string to be applied to the check timestamp

  • interval_timezone – timezone string to be applied to the interval timestamps

utils.functions.decode_base64_file(source_filename: str, destination_filename: str)

Decodes a base64 file while it’s read (no memory allocation). Suitable for big file conversion.

Parameters:

  • source_filename – source base64 encoded file

  • destination_filename – destination decoded file

utils.functions.deep_get_attribute(attribute: str, obj: any, values: list[any] | None = None) → list[any]

Recursively finds the attribute value in all the occurrences of the object and returns the value.

Parameters:

  • attribute – the attribute to search in the object

  • obj – the object that will be traversed and modified

  • values – the list of values found each step that will be recursively passed to the function itself each time. can be left empty.

Returns:

a list of values found

utils.functions.deep_set_attribute(attribute: str, value: str, obj: any) → any

Recursively sets the attribute value in all the occurrences of the object. An example usage is to anonymize a yaml object setting all the occurrences of the property kubeconfig with a dummy value.

Parameters:

  • attribute – the attribute name in the object

  • value – the value that will be set in the attribute if present

  • obj – the object that will be traversed and modified

Returns:

obj

utils.functions.filter_dictionary(datetime_key: str, start_timestamp: int | None, end_timestamp: int | None, dictionary_timezone: str, interval_timezone: str) → dict[str, any] | None

Filters a dictionary by datetime string attribute

Parameters:

  • dictionary – the dictionary that needs to be filtered

  • datetime_key – the key of the dictionary representing the time of the dictionary

  • start_timestamp – timestamp representing the minimum date after that the dictionary becomes relevant, if None no bottom limit applied

  • end_timestamp – timestamp representing the maximum date before that the dictionary is still relevant, if None no top limit applied

  • dictionary_timezone – timezone of the date contained in the dictionary

  • interval_timezone – timezone of the interval within the dictionary will be checked

utils.functions.filter_log_file_worker(start_timestamp: int | None, end_timestamp: int | None, src_folder: str, dst_folder: str, remote_timezone: str, local_timezone: str, log_filter_patterns: list[str], queue: Queue)

Log file filter worker. Filters a file scanning each line (naive approach, no algorithms impletemented for this first version) and extracting time infos matching it with the provided regular expression and time range.

Parameters:

  • start_timestamp – timestamp of the first relevant entry, if None will start filter starting from the earliest

  • end_timestamp – timestamp of the last relevant entry, if None will end filtering until the latest

  • src_folder – used to remove from the final filtered log name the base directory that is not relevant

  • dst_folder – output folder where the filtered file will be placed

  • local_timezone – timezone of the client

  • remote_timezone – timezone of the system from which the logs have been extracted

  • log_filter_patterns – a list of regex that will match and extract the time info that will be parsed by dateutil.parser (it supports several formats by default but not every date format). Each pattern must contain only 1 group that represent the time string that must be extracted and parsed

  • queue – a queue containing pathlib.Path objects representing the log file to be parsed

utils.functions.filter_log_line(log_line: str, start_timestamp: int | None, end_timestamp: int | None, remote_timezone: str, local_timezone: str, log_filter_patterns: [re.Pattern[str]]) → str | None

Filters a log line extracting time informations using a set of compiled regular expressions containing excatly one group.

Parameters:

  • start_timestamp – timestamp representing the minimum date after that the log becomes relevant, if None no bottom limit applied

  • end_timestamp – timestamp representing the maximum date before that the log is still relevant, if None no top limit applied

  • remote_timezone – timezone of the system from where the log has been extracted

  • local_timezone – timezone of the client that is parsing the log

  • log_filter_patterns – a list of regex that will match and extract the time info that will be parsed by dateutil.parser (it supports several formats by default but not every date format). Each pattern must contain only 1 group that represent the time string that must be extracted and parsed

Returns:

the log line if matches the criteria above otherwise None

utils.functions.find_executable_in_path(executable_name: str) → str | None
utils.functions.get_junit_test_case(success: bool, time: int, test_suite_name: str, test_case_description: str, test_stdout: str = '', test_version: str | None = None) → str

Creates an XML compatible with sippy to track regressions on OCP tests.

Parameters:

  • success – if true will print a success test case, otherwise a failure test case

  • time – sets the duration in seconds of the testsuite

  • test_suite_name – sets the name of the test suite

  • test_case_description – sets the description of the testcase, it has to contain tags that map the test case to the monitored component on sippy like [sig-etcd] or others

  • test_stdout – if a test failes the stdout of the krkn-run is attached to the testcase element in the XML

  • test_version – sets an optional property to the testsuite containing the version on the test

Returns:

the XML string to be written in the junit xml file

utils.functions.get_random_string(length: int) → str

Returns a random lowercase string of lenght length

Parameters:

length – the lenght of the string

Returns:

the random string

utils.functions.get_yaml_item_value(cont: dict[str, any], item: str, default: any) → any

Sets the value of item from yaml.

Parameters:

  • cont – dict of all items from yaml file

  • item – name of the item in scenario yaml

  • default – default value

Returns:

item value - if not specified the default value is returned

utils.functions.is_host_reachable(host: str, port: int, timeout: int = 2) → bool
utils.functions.log_exception(scenario: str | None = None)

Logs an exception printing the file and the line number from where the method is called

Parameters:

scenario – if set will include the scenario name in the log