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.deploy module

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

ocs_ci.framework.exceptions module

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

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.ClusterNotAccessibleError: Bases: Exception

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

exception ocs_ci.framework.exceptions.InvalidDeploymentType: 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.get_step_count(module=None)

Get current step count.

Parameters:: module – None returns dict copy; else module object or name.

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.logger_helper.reset_current_module_log_steps(): Reset step count for the module from which this helper is called. Call this function f.e. if log_step is used in a loop and we want to refresh counting on each iteration.

ocs_ci.framework.logger_helper.reset_log_steps(module=None)

Reset step counters.

Parameters:

module –

None: clear all
module object or module name string: clear only that module

ocs_ci.framework.main module

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

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>, PREUPGRADE_CONFIG: dict = <factory>)

Bases: object

AUTH: dict

COMPONENTS: dict

DEPLOYMENT: dict

ENV_DATA: dict

EXTERNAL_MODE: dict

FLEXY: dict

MULTICLUSTER: dict

PERF: dict

PREUPGRADE_CONFIG: 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.ConfigSafeThread(config_index, *args, **kwargs)

Bases: Thread

This is customized threading.Thread which is safe to use within our framework with config object. This ConfigSafeThread prevents a situation where one thread changes its context of config and modifies other running thread (e.g. main thread of framework) config context. The instance of ConfigSafeThread will define config index which will be used by all the calls in the thread. It uses config.thread_local_data with specific Thread ID and config ID to be used by the thread for its life cycle.

run(*args, **kwargs)

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

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 RunWithFirstConsumerConfigContextIfAvailable

Bases: RunWithConfigContext

Context manager that makes sure that a given code block is executed on First consumer. If Consumer config is not available then run with current config context.

class RunWithPrimaryConfigContext: Bases: RunWithConfigContext

class RunWithProviderConfigContextIfAvailable

Bases: RunWithConfigContext

Context manager that makes sure that a given code block is executed on Provider. If Provider config is not available then run with current config context.

property cluster_ctx

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

property default_cluster_index

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

Returns:: The default cluster context index
Return type:: int

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_client_contexts_if_available(): Get contexts that can be used for context iteration of client clusters. If there are no client contexts available then use simple nullcontext to not break functionality and still execute the code on current cluster.

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_kubeconfig_by_index(index)

Get the cluster kubeconfig by the cluster index

Parameters:: index (int) – The cluster index
Returns:: The cluster kubeconfig path
Return type:: str
Raises:: ClusterNotFoundException – In case it didn’t find the cluster

get_cluster_name_by_index(index)

Get the cluster name by the cluster index

Parameters:: index (int) – The cluster index
Returns:: The cluster name
Return type:: str
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(raise_exception=True)

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

Get the provider cluster indexes

Returns:: The indexes of provider clusters
Return type:: list

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

insert_cluster_config(index, new_config)

Insert a new cluster configuration at the given index

Parameters:

index (int) – The index at which to insert the new configuration
new_config (Config) – The new configuration to insert

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

remove_cluster(index)

Remove the cluster at the given index

Parameters:: index (int) – The index of the cluster to remove

remove_cluster_by_name(cluster_name)

Remove the cluster by the cluster name

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

reset()

reset_ctx()

run_for_all_clusters(func): A decorator to run the decorated function for all clusters and switch context between them.

static run_with_provider_context_if_available(func)

Decorator that runs the function using the Provider config if it exists. If no Provider config is found, the function runs with the current config.

Parameters:: func (callable) – Function to decorate.
Returns:: Wrapped function.
Return type:: callable

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.config_safe_thread_pool_task(config_index, task, *args, **kwargs)

Wrapper function to be executed in ThreadPoolExecutor

Parameters:

config_index (int) – first positional argument defining config index to be used by Thread.
task (function) – function to be called by ThreadPoolExecutor

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',
}