ocs_ci.framework package

Subpackages

Submodules

ocs_ci.framework.custom_logger module

Custom logger for OCS-CI framework.

Provides enhanced logging with custom levels for test automation: - TEST_STEP: Sequential test steps with automatic numbering - ASSERTION: Test assertions and validations - AI_DATA: AI/ML metrics, predictions, and analysis

This module patches Python’s logging module via logging.setLoggerClass() so ALL calls to logging.getLogger() automatically return OCSCILogger instances.

Usage:

import logging # Standard import - no changes needed!

logger = logging.getLogger(__name__) # Returns OCSCILogger automatically logger.test_step(“Create PVC”) logger.assertion(“PVC status: expected=’Bound’, actual=’Bound’”) logger.ai_data(“Failure prediction: 85% probability”)

class ocs_ci.framework.custom_logger.OCSCILogger(name, level=0)

Bases: Logger

Custom logger for OCS-CI framework.

Automatically used by all logging.getLogger() calls via logging.setLoggerClass().

Provides custom log levels and methods: - test_step(): Log test steps with automatic numbering - assertion(): Log test assertions and validations - ai_data(): Log AI/ML metrics and predictions

ai_data(message, *args, **kwargs)

Log AI/ML data at AI_DATA level.

This level is below DEBUG (5 < 10) and requires explicit enabling. Use for AI/ML metrics, predictions, model information, and analysis data. To see AI_DATA logs, set log level to 5 or use –log-cli-level=5

Parameters:

message (str) – AI/ML data description
*args – Format arguments for message
**kwargs – Additional logging kwargs (exc_info, extra, etc.)

Example

logger.ai_data(“Prediction: failure_risk=0.85, model=’v2.3’, confidence=0.95”)

assertion(message, *args, **kwargs)

Log assertion at ASSERTION level.

Use for test validations and assertions to make them easily identifiable in logs and filterable separately from regular INFO messages.

Parameters:

message (str) – Assertion description
*args – Format arguments for message
**kwargs – Additional logging kwargs (exc_info, extra, etc.)

Example

logger.assertion(“PVC status: expected=’Bound’, actual=’Bound’”)

test_step(message, *args, **kwargs)

Log test step at TEST_STEP level with automatic numbering.

The step number is automatically incremented per module and includes the calling function name in the format: “function_name — N — message”

Parameters:

message (str) – Step description
*args – Format arguments for message
**kwargs – Additional logging kwargs (exc_info, extra, etc.)

Example

logger.test_step(“Create PVC”) # Output: “test_pvc_creation — 1 — Create PVC”

ocs_ci.framework.custom_logger.get_current_step(module_name)

Get current step count for a module without incrementing.

Parameters:: module_name (str) – Name of the module/logger
Returns:: Current step number (0 if not yet incremented)
Return type:: int

ocs_ci.framework.custom_logger.increment_step(module_name)

Atomically increment and return step number for a module.

Parameters:: module_name (str) – Name of the module/logger
Returns:: The new step number
Return type:: int

ocs_ci.framework.custom_logger.reset_step_counts(module_name=None)

Reset step counters.

Parameters:: module_name (str, optional) – Name of specific module to reset. If None, resets all counters.

ocs_ci.framework.exceptions module

exception ocs_ci.framework.exceptions.ClusterNameLengthError(name, min_length=5, max_length=17): Bases: Exception

exception ocs_ci.framework.exceptions.ClusterNameNotProvidedError: Bases: Exception

exception ocs_ci.framework.exceptions.ClusterPathNotProvidedError: Bases: Exception

ocs_ci.framework.logger_factory module

ocs_ci.framework.logger_factory.record_factory(*args, **kwargs): Record factory setup function :returns: Reference obj to the record of logger :rtype: logging.record

ocs_ci.framework.logger_factory.set_log_record_factory(): Custom attribute additions to logging are addressed in this function Override the logging format with a new log record factory :returns: None

ocs_ci.framework.logger_helper module

ocs_ci.framework.logger_helper.log_step(message: str)

Method to log step in the test case.

Deprecated since version Use: logger.test_step() instead. The custom logger now provides test_step() method on all loggers automatically::

logger = logging.getLogger(__name__) logger.test_step(“Your step message”)

Log will be in the format: <test_function_name> — <step_number> — <message> Similar to: odf_overview_ui — 8 — Navigate Storage System via breadcrumb

Parameters:: message (str) – Message to be logged

ocs_ci.framework.main module

ocs_ci.framework.main.check_config_requirements()

Checking if all required parameters were passed

Raises:: MissingRequiredConfigKeyError – In case of some required parameter is not defined.

ocs_ci.framework.main.init_multicluster_ocsci_conf(args, nclusters)

Parse multicluster specific arguments and seperate out each cluster’s configuration. Then instantiate Config class for each cluster

Params:: args (list): of arguments passed nclusters (int): Number of clusters (>1)

ocs_ci.framework.main.init_ocsci_conf(arguments=None)

Update the config object with any files passed via the CLI

Parameters:: arguments (list) – Arguments for pytest execution

ocs_ci.framework.main.load_config(config_files)

This function load the config files in the order defined in config_files list.

Parameters:: config_files (list) – config file paths

ocs_ci.framework.main.main(argv=None)

ocs_ci.framework.main.process_ocsci_conf(arguments)

ocs_ci.framework.main.signal_term_handler(sig, frame)

ocs_ci.framework.main.tokenize_per_cluster_args(args, nclusters)

Seperate per cluster arguments so that parsing becomes easy

Params:: args: Combined arguments nclusters(int): total number of clusters

Returns:

Each cluster conf per list: ex: [[cluster1_conf], [cluster2_conf]…]

Return type:

list of lists

ocs_ci.framework.testlib module

class ocs_ci.framework.testlib.BaseTest

Bases: object

Base test class for our testing. If some functionality/property needs to be implemented in all test classes here is the place to put your code.

pytestmark = [Mark(name='usefixtures', args=('resource_checker',), kwargs={}), Mark(name='usefixtures', args=('environment_checker',), kwargs={})]

class ocs_ci.framework.testlib.E2ETest

Bases: BaseTest

Base class for E2E team

pytestmark = [Mark(name='usefixtures', args=('resource_checker',), kwargs={}), Mark(name='usefixtures', args=('environment_checker',), kwargs={}), Mark(name='e2e', args=(), kwargs={})]

class ocs_ci.framework.testlib.EcosystemTest

Bases: BaseTest

Base class for E2E team

pytestmark = [Mark(name='usefixtures', args=('resource_checker',), kwargs={}), Mark(name='usefixtures', args=('environment_checker',), kwargs={}), Mark(name='ecosystem', args=(), kwargs={})]

class ocs_ci.framework.testlib.MCGTest

Bases: ManageTest

MAX_ENDPOINT_COUNT = 2

MIN_ENDPOINT_COUNT = 2

pytestmark = [Mark(name='usefixtures', args=('resource_checker',), kwargs={}), Mark(name='usefixtures', args=('environment_checker',), kwargs={}), Mark(name='manage', args=(), kwargs={}), Mark(name='usefixtures', args=('nb_ensure_endpoint_count',), kwargs={}), Mark(name='ignore_leftover_label', args=('noobaa-s3=noobaa',), kwargs={})]

class ocs_ci.framework.testlib.ManageTest

Bases: BaseTest

Base class for E2E team

pytestmark = [Mark(name='usefixtures', args=('resource_checker',), kwargs={}), Mark(name='usefixtures', args=('environment_checker',), kwargs={}), Mark(name='manage', args=(), kwargs={})]

Module contents

Avoid already-imported warning cause of we are importing this package from run wrapper for loading config.

You can see documentation here: https://docs.pytest.org/en/latest/reference.html under section PYTEST_DONT_REWRITE

class ocs_ci.framework.Config(AUTH: dict = <factory>, DEPLOYMENT: dict = <factory>, ENV_DATA: dict = <factory>, EXTERNAL_MODE: dict = <factory>, REPORTING: dict = <factory>, RUN: dict = <factory>, UPGRADE: dict = <factory>, FLEXY: dict = <factory>, UI_SELENIUM: dict = <factory>, PERF: dict = <factory>, COMPONENTS: dict = <factory>, MULTICLUSTER: dict = <factory>)

Bases: object

AUTH: dict

COMPONENTS: dict

DEPLOYMENT: dict

ENV_DATA: dict

EXTERNAL_MODE: dict

FLEXY: dict

MULTICLUSTER: dict

PERF: dict

REPORTING: dict

RUN: dict

UI_SELENIUM: dict

UPGRADE: dict

get_defaults(): Return a fresh copy of the default configuration

reset(): Clear all configuration data and load defaults

to_dict()

update(user_dict: dict): Override configuration items with items in user_dict, without wiping out non-overridden items

class ocs_ci.framework.GlobalVariables

Bases: object

TIMEREPORT_DICT: dict = {}

class ocs_ci.framework.MultiClusterConfig

Bases: object

class RunWithAcmConfigContext: Bases: RunWithConfigContext

class RunWithConfigContext(config_index): Bases: object

class RunWithPrimaryConfigContext: Bases: RunWithConfigContext

attr_init()

current_cluster_name()

Get the Cluster name of the current context

Returns:: The cluster name which is stored as str in config (None if key not exist)
Return type:: str

property default_cluster_ctx

Get the default cluster context. The default cluster context will be defined by the default index as defined in the ‘ENV DATA’ param ‘default_cluster_context_index’

Returns:: The default cluster context
Return type:: ocs_ci.framework.Config

designate_active_acm_cluster()

Set one of the ACM clusters as the active ACM cluster. This is done in the event that none of the ACM clusters are set as active.

Returns:: The multicluster index of the newly designated active ACM cluster
Return type:: int

get_active_acm_index()

Retrieve the active ACM cluster index.

Returns:: The multicluster_index of the active ACM cluster config.
Return type:: int

get_cluster_index_by_name(cluster_name)

Get the cluster index by the cluster name

Returns:: The cluster index by the cluster name
Return type:: int
Raises:: ClusterNotFoundException – In case it didn’t find the cluster

get_cluster_type_indices_list(cluster_type)

Get the cluster type indices

Returns:: the cluster type indices
Return type:: list
Raises:: ClusterNotFoundException – In case it didn’t find any cluster with the cluster type

get_consumer_indexes_list()

Get the consumer cluster indexes

Returns:: the consumer cluster indexes
Return type:: list
Raises:: ClusterNotFoundException – In case it didn’t find any consumer cluster

get_defaults()

get_provider_index()

Get the provider cluster index

Returns:: the provider cluster index
Return type:: int
Raises:: ClusterNotFoundException – In case it didn’t find the provider cluster

hci_client_exist()

Check if the hci_client cluster exists in the clusters

Returns:: True, if the hci_client cluster exists in the clusters. False, otherwise.
Return type:: bool

hci_provider_exist()

Check if the provider cluster exists in the clusters

Returns:: True, if the provider cluster exists in the clusters. False, otherwise.
Return type:: bool

init_cluster_configs()

is_cluster_type_exist(cluster_type)

Check if the given cluster type exists in the clusters

Parameters:: cluster_type (str) – The cluster type
Returns:: True, if the given cluster type exists in the clusters. False, otherwise.
Return type:: bool

is_consumer_exist()

Check if the consumer cluster exists in the clusters

Returns:: True, if the consumer cluster exists in the clusters. False, otherwise.
Return type:: bool

is_provider_exist()

Check if the provider cluster exists in the clusters

Returns:: True, if the provider cluster exists in the clusters. False, otherwise.
Return type:: bool

reset()

reset_ctx()

switch_acm_ctx()

switch_ctx(index=0)

switch_default_cluster_ctx()

switch_to_cluster_by_cluster_type(cluster_type, num_of_cluster=0)

Switch to the cluster with the given cluster type

Parameters:

cluster_type (str) – The cluster type
num_of_cluster (int) – The cluster index to switch to. The default cluster number is 0 - which means it will switch to the first cluster. 1 - is the second, 2 - is the third, and so on.

Raises:

ClusterNotFoundException – In case it didn’t find any cluster with the cluster type

switch_to_cluster_by_name(cluster_name)

Switch to the cluster by the cluster name

Parameters:: cluster_name (str) – The cluster name to switch to
Raises:: ClusterNotFoundException – In case it didn’t find the cluster

switch_to_consumer(num_of_consumer=0)

Switch to one of the consumer clusters

Parameters:: num_of_consumer (int) – The cluster index to switch to. The default consumer number is 0 - which means it will switch to the first consumer. 1 - is the second, 2 - is the third, and so on.
Raises:: ClusterNotFoundException – In case it didn’t find the consumer cluster

switch_to_provider()

Switch to the provider cluster

Raises:: ClusterNotFoundException – In case it didn’t find the provider cluster

update(user_dict)

ocs_ci.framework.merge_dict(orig: dict, new: dict) → dict

Update a dict recursively, with values from ‘new’ being merged into ‘orig’.

Parameters:

orig (dict) – The object that will receive the update
new (dict) – The object which is the source of the update

Example:

orig = {
    'dict': {'one': 1, 'two': 2},
    'list': [1, 2],
    'string': 's',
}
new = {
    'dict': {'one': 'one', 'three': 3},
    'list': [0],
    'string': 'x',
}
merge_dict(orig, new) ->
{
    'dict': {'one': 'one', 'two': 2, 'three': 3}
    'list': [0],
    'string', 'x',
}