ocs_ci.helpers package

Submodules

ocs_ci.helpers.bdi_helpers module

Helper functions specific for bdi

ocs_ci.helpers.bdi_helpers.clean_temp_files(class_instance): Deletes temporary files created during the execution

ocs_ci.helpers.bdi_helpers.init_configure_workload_yaml(namespace=None, image=None, sf=1, pvc_name=None)

Loads the “configure-workload.yaml” and sets all the relevant parameters according to the type of execution

Parameters:

namespace (str) – The name of the namespace
image (str) – The name of the image to pull
sf (int) – scale factor. Can be 1, 10, 100…
pvc_name (str) – The name of the pvc to be used

Returns:

The dictionary representing the yaml with all the relevant changes

Return type:

dict

ocs_ci.helpers.bdi_helpers.init_data_load_yaml(namespace=None, image=None, pvc_name=None, secret_name=None)

Loads the “data-load-job” and sets all the relevant parameters according to the type of execution

Parameters:

namespace (str) – The name of the namespace
image (str) – The name of the image to pull
pvc_name (str) – The name of the pvc to be used
secret_name (str) – The secret name to be used

Returns:

The dictionary representing the yaml with all the relevant changes

Return type:

dict

ocs_ci.helpers.bdi_helpers.init_run_workload_yaml(namespace=None, image=None, pvc_name=None, secret_name=None)

Loads the “run-workload-job” and sets all the relevant parameters according to the type of execution

Parameters:

namespace (str) – The name of the namespace
image (str) – The name of the image to pull
pvc_name (str) – The name of the pvc to be used
secret_name (str) – The secret name to be used

Returns:

The dictionary representing the yaml with all the relevant changes

Return type:

dict

ocs_ci.helpers.bdi_helpers.install_db2u(class_instance): Chart installation

ocs_ci.helpers.ceph_helpers module

ocs_ci.helpers.ceph_helpers.get_mon_quorum_count() → int

Get the current number of monitors in quorum.

Returns:: The number of monitors currently in quorum.
Return type:: int

ocs_ci.helpers.ceph_helpers.get_mon_quorum_ranks() → dict

Returns a mapping of mon names to ranks using ‘ceph mon stat’. Only monitors currently in quorum are returned.

Returns:: A dictionary mapping monitor names to their ranks.
Return type:: dict

ocs_ci.helpers.ceph_helpers.get_mon_status(mon_id: str) → str

Check the status of a monitor by verifying its presence in the quorum.

Parameters:: mon_id – The monitor ID (e.g., ‘a’, ‘b’, ‘c’) to check
Returns:: MON_STATUS_UP if the monitor is in quorum, MON_STATUS_DOWN otherwise.
Return type:: str

ocs_ci.helpers.ceph_helpers.wait_for_ceph_used_capacity_reached(expected_used_capacity, timeout=1800, sleep=20)

Wait until the cluster used Ceph capacity in GiB reaches or exceeds a specified threshold.

Parameters:

expected_used_capacity (int|float) – The used capacity in GiB to wait for.
timeout (int) – Maximum time to wait in seconds. Defaults to 1800 seconds (30 minutes).
sleep (int) – Time to wait between checks in seconds. Defaults to 20 seconds.

Raises:

TimeoutExpiredError – If the expected capacity is not reached within the timeout.

ocs_ci.helpers.ceph_helpers.wait_for_mon_status(mon_id: str, status: str = 'up', timeout: int = 300, sleep: int = 20) → None

Wait until a specified monitor reaches the desired status

Parameters:

mon_id – The monitor ID (e.g., ‘a’, ‘b’, ‘c’) to check
status – The desired status to wait for (‘up’ or ‘down’). Defaults to ‘up’.
timeout – Maximum time to wait in seconds. Defaults to 300 seconds.
sleep – Time to wait between checks in seconds. Defaults to 20 seconds.

Raises:

TimeoutExpiredError – If the monitor does not reach the desired status within the timeout.

ocs_ci.helpers.ceph_helpers.wait_for_mons_in_quorum(expected_mon_count, timeout=300, sleep=20) → None

Wait until the number of monitors in quorum reaches the expected count.

Parameters:

expected_mon_count (int) – The expected number of monitors in quorum.
timeout (int) – Maximum time to wait in seconds. Defaults to 300 seconds (5 minutes).
sleep (int) – Time to wait between checks in seconds. Defaults to 10 seconds.

Raises:

TimeoutExpiredError – If the expected number of monitors in quorum is not reached within the timeout.

ocs_ci.helpers.ceph_helpers.wait_for_percent_used_capacity_reached(expected_used_capacity, timeout=1800, sleep=20)

Wait until the used capacity percentage reaches or exceeds a specified threshold.

This function repeatedly samples the current used capacity using get_percent_used_capacity() until it meets or exceeds the expected_used_capacity or until the timeout is reached.

Parameters:

expected_used_capacity (int or float) – The percentage of used capacity to wait for.
timeout (int) – Maximum time to wait in seconds. Defaults to 1800 seconds (30 minutes).
sleep (int) – Time to wait between checks in seconds. Defaults to 20 seconds.

Raises:

TimeoutExpiredError – If the expected capacity is not reached within the timeout.

ocs_ci.helpers.ceph_mon_healthcheck module

ocs_ci.helpers.ceph_mon_healthcheck.delete_storagecluster_mon_healthcheck(sc_obj=None) → bool

Delete the ceph mon healthcheck configuration from the storagecluster resource

Parameters:: sc_obj (ocs_ci.ocs.ocp.OCP) – StorageCluster object. If None, it will be fetched.
Returns:: True if the deletion was successful, False otherwise
Return type:: bool

ocs_ci.helpers.ceph_mon_healthcheck.extract_timeout_seconds(line: str) → int | None

Extract the timeout seconds value from a log line. :param line: A log line containing the timeout message. :type line: str

Returns:: The extracted timeout seconds value, or None if not found.
Return type:: int | None

ocs_ci.helpers.ceph_mon_healthcheck.get_cephcluster_mon_healthcheck(cc_obj=None) → dict

Get the ceph mon healthcheck status from the cephcluster resource Handles both object structures: - cc_obj[“items”][0][“spec”][“healthCheck”][“daemonHealth”][“mon”] - cc_obj[“spec”][“healthCheck”][“daemonHealth”][“mon”]

Parameters:: cc_obj (object) – CephCluster object. If None, it will be fetched.
Returns:: The mon healthCheck configuration
Return type:: dict

ocs_ci.helpers.ceph_mon_healthcheck.get_storagecluster_mon_healthcheck(sc_obj=None) → dict

Get the ceph mon healthcheck status from the storagecluster resource Handles both object structures: - sc_obj[“items”][0][“spec”][“managedResources”][“cephCluster”][“healthCheck”][“daemonHealth”][“mon”] - sc_obj[“spec”][“managedResources”][“cephCluster”][“healthCheck”][“daemonHealth”][“mon”]

Parameters:: sc_obj (ocs_ci.ocs.ocp.OCP) – StorageCluster object. If None, it will be fetched.
Returns:: The mon healthCheck configuration
Return type:: dict

ocs_ci.helpers.ceph_mon_healthcheck.patch_storagecluster_mon_healthcheck(mon_timeout, mon_interval='20s', sc_obj=None) → bool

Patch the ceph mon healthcheck configuration in the storagecluster resource

Parameters:

mon_timeout (str) – The timeout value for the mon healthcheck(e.g., “3m”, “20m”, 600s”)
mon_interval (str) – The interval value for the mon healthcheck (e.g., “20s”, “1m”)
sc_obj (ocs_ci.ocs.ocp.OCP) – StorageCluster. If None, it will be fetched.

Returns:

True if the patch was successful, False otherwise

Return type:

bool

ocs_ci.helpers.ceph_mon_healthcheck.select_mon_id_and_node() → tuple[str, str]

Select a monitor ID and its corresponding node name.

Returns:: A tuple containing the monitor ID (str) and node name (str).
Return type:: tuple

ocs_ci.helpers.ceph_mon_healthcheck.verify_mon_healthcheck_consistency(sc_obj=None, cc_obj=None) → bool

Verify that the mon healthcheck configurations in StorageCluster and CephCluster are consistent.

Parameters:

sc_obj (ocs_ci.ocs.ocp.OCP) – StorageCluster object. If None, it will be fetched.
cc_obj (object) – CephCluster object. If None, it will be fetched.

Returns:

True if configurations match, False otherwise

Return type:

bool

ocs_ci.helpers.ceph_mon_healthcheck.verify_mon_healthcheck_timeout_value_in_logs(mon_id, timeout_value, since='1m', timeout=300, sleep=20) → bool

Verify that the expected mon healthcheck timeout value appears in the rook-ceph-operator pod logs. The function waits for the log pattern to appear and then checks if the timeout value is present in the matched log lines.

Parameters:

mon_id (str) – The monitor ID to check in the logs.
timeout_value (int) – The timeout value to check in the logs.
since (str) – Time duration to look back in the logs.
timeout (int) – Maximum time to wait for the patterns.
sleep (int) – Time to wait between log checks.

Returns:

True if the timeout value is found, False otherwise.

Return type:

bool

ocs_ci.helpers.ceph_mon_healthcheck.wait_for_mon_healthcheck_consistency(sc_obj=None, cc_obj=None, timeout=120, sleep=10) → None

Wait until the mon healthcheck configurations in StorageCluster and CephCluster are consistent.

Parameters:

sc_obj (ocs_ci.ocs.ocp.OCP) – StorageCluster object. If None, it will be fetched.
cc_obj (object) – CephCluster object. If None, it will be fetched.
timeout (int) – Maximum time to wait in seconds. Defaults to 300 seconds (5 minutes).
sleep (int) – Time to wait between checks in seconds. Defaults to 20 seconds.

Raises:

TimeoutExpiredError – If the configurations do not match within the timeout.

ocs_ci.helpers.ceph_mon_healthcheck.wait_for_mon_healthcheck_timeout_in_logs(mon_id, since='1m', timeout=300, sleep=20) → list

Wait for expected mon healthcheck timeout patterns to appear in the rook-ceph-operator pod logs and return the matched log lines.

Parameters:

mon_id (str) – The monitor ID (e.g., ‘a’, ‘b’, ‘c’) to check in the logs.
since (str) – Time duration to look back in the logs (e.g., ‘1m’, ‘5m’).
timeout (int) – Maximum time in seconds to wait for the patterns to appear.
sleep (int) – Interval in seconds to wait between log polling attempts.

Returns:

A list of matched log lines containing the timeout message.

Return type:

list

Raises:

ValueError – If the rook-ceph-operator pod is not found.
TimeoutExpiredError – If no matching pattern is found within the timeout period.

Example

>>> wait_for_mon_healthcheck_timeout_in_logs("c", since="2m", timeout=600, sleep=30)
[
    "2025-11-25 13:08:47... mon 'c' not found in quorum, waiting for timeout (599 seconds left)...",
    "2025-11-25 13:09:32... mon 'c' not found in quorum, waiting for timeout (554 seconds left)...",
]

ocs_ci.helpers.ceph_mon_healthcheck.wait_for_mon_pod_restart(mon_id, timeout=300, sleep=20) → None

Wait until the monitor pod for a specified monitor ID restarts.

Parameters:

mon_id (str) – The monitor ID (e.g., ‘a’, ‘b’, ‘c’) to wait for.
timeout (int) – Maximum time to wait in seconds. Defaults to 300 seconds (5 minutes).
sleep (int) – Time to wait between checks in seconds. Defaults to 20 seconds.

Raises:

TimeoutExpiredError – If the monitor pod does not restart within the timeout.

ocs_ci.helpers.cephfs_stress_helpers module

CephFS Stress Test Helper Module

This module provides comprehensive utilities for managing CephFS stress tests, including pod/job creation, background health monitoring, cluster verification and resource cleanup.

class ocs_ci.helpers.cephfs_stress_helpers.CephFSStressTestManager(namespace)

Bases: object

Manages CephFS stress test lifecycle with pod/job creation, background health monitoring and resource cleanup.

create_cephfs_stress_job(pvc_name, base_dir=None, files_size=None, operations=None, base_file_count=None, multiplication_factors=None, threads=None, parallelism=None, completions=None)

Creates a CephFS stress Job. This job launches concurrent pods based on the configured parallelism count, where each pod executes generate numerous small files and directories. Configured with specific parameters, the workload stresses CephFS by gradually increasing the load in incremental stages.

Parameters:

pvc_name (str) – Name of the PersistentVolumeClaim
base_dir (str, optional) – Directory used by smallfile to perform file and directory operations
files_size (str, optional) – Size of each file in KB
operations (str, optional) – File operations to perform (e.g., append, stat, chmod, ls-l, etc),
string (Pass as a comma-separated) –
base_file_count (str, optional) – Base file count, to multiply with scaling factor
multiplication_factors (str, optional) – Dynamic scaling of file creation
MULTIPLICATION_FACTORS (- base_file_count *) –
threads (str, optional) – Number of threads to use for the operation.
parallelism (str, optional) – Specifies how many pod replicas running in parallel should execute a job.
completions (str, optional) – Specifies how many times the Pod must finish successfully before the entire
"Complete. (Job is marked as) –

Returns:

The created Job object after it’s in a running state

Return type:

cephfs_stress_job_obj(OCS)

Raises:

AssertionError – If the Job creation fails

create_cephfs_stress_pod(pvc_name, base_dir=None, files_size=None, operations=None, base_file_count=None, multiplication_factors=None, threads=None)

Creates a CephFS stress pod, utilizing smallfiles to generate numerous files and directories.

The pod is configured with various parameters to stress CephFS and gradually increases load in incremental stages.

Parameters:

pvc_name (str) – Name of the PersistentVolumeClaim
base_dir (str, optional) – Directory used by smallfile to perform file and directory operations
files_size (str, optional) – Size of each file in KB
operations (str, optional) – File operations to perform (e.g., append, stat, chmod, ls-l, etc),
string (Pass as a comma-separated) –
base_file_count (str, optional) – Base file count, to multiply with scaling factor
multiplication_factors (str, optional) – Dynamic scaling of file creation
MULTIPLICATION_FACTORS (- base_file_count *) –
threads (str, optional) – Number of threads to use for the operation.

Returns:

The created Pod object after it’s in a running state

Return type:

pod_obj

Raises:

AssertionError – If the pod creation fails

pause_background_checks()

Pause background verification checks temporarily.

This is useful during intentional disruptions (e.g., node failures, pod restarts) where verification failures are expected and should not fail the test.

resume_background_checks()

Resume background verification checks after they were paused.

Should be called after cluster has recovered from intentional disruptions and is expected to be in a healthy state.

setup_stress_test_environment(pvc_size)

Creates the foundational resources (PVC and Standby Pod) for the stress test.

Parameters:: pvc_size (str) – Size of pvc to create
Returns:: Created PVC and standby pod objs
Return type:: tuple

start_background_checks(interval_minutes=5)

Starts the background thread (‘StressWatchdog’) for continuous cluster monitoring.

The background thread runs periodic health checks and verifications at the specified interval. If the thread is already running, this method returns without creating a new thread.

Parameters:: interval_minutes – Interval in minutes between check executions

stop_background_checks(timeout=10): Signals the background thread (‘StressWatchdog’) to stop and waits for it to join.

teardown(): Stops background checks, collects output directory, and deletes created resources.

ocs_ci.helpers.cephfs_stress_helpers.check_ceph_health(stress_manager=None)

Checks the health of the Ceph cluster.

Parameters:: stress_manager – CephFSStressTestManager instance to check pause status
Raises:: Exception – If Ceph cluster is not healthy

ocs_ci.helpers.cephfs_stress_helpers.check_for_filesystem_hangs(namespace, output_dir='/mnt/output')

Check for filesystem hang markers created by the monitoring script.

This function checks all pods in the given namespace for hang marker files that indicate the filesystem monitoring detected a genuine hang.

Parameters:

namespace (str) – Namespace to check for hang markers
output_dir (str) – Output directory path where hang markers are stored

Returns:

(hang_detected: bool, hang_details: list of dicts)

Return type:

tuple

Raises:

Exception – If hang markers are found (genuine filesystem hang detected)

ocs_ci.helpers.cephfs_stress_helpers.check_mds_pods_resource_utilization(): Gets the current resource utilization of MDS pods from ‘adm top’ command.

ocs_ci.helpers.cephfs_stress_helpers.check_prometheus_alerts(stress_manager)

Fetches alerts from the PrometheusAPI and logs alerts in a tabulated format

Parameters:: stress_manager – CephFSStressTestManager instance with prometheus_api

ocs_ci.helpers.cephfs_stress_helpers.collect_monitoring_logs(stress_job_obj, dir_name=None)

Collect filesystem monitoring logs from stress job pods.

Parameters:

stress_job_obj – Stress job object whose monitoring logs need to be collected
dir_name (str) – Optional subdirectory name for organizing logs

ocs_ci.helpers.cephfs_stress_helpers.collect_stress_job_output_directory(standby_pod_obj, dir_name=None)

Collect entire output directory from shared CephFS mount and store in ocs-ci log directory. This collects all files including monitoring logs, hang markers, and test artifacts.

Uses the standby pod (which is always running) to access the shared CephFS mount, avoiding issues with completed job pods that can’t execute commands.

Parameters:

standby_pod_obj – Standby pod object that mounts the shared PVC (must be running)
dir_name (str) – Optional subdirectory name. By default files are stored in ocs-ci-logs-<run_id>/<test_name>/stress_output directory. When dir_name is provided, files are stored in ocs-ci-logs-<run_id>/<test_name>/stress_output/<dir_name>

ocs_ci.helpers.cephfs_stress_helpers.collect_stress_job_pod_logs(stress_job_obj, dir_name=None)

Collect stress job pod logs and store them in ocs-ci log directory.

Parameters:

stress_job_obj – Stress job object whose pod logs need to be collected
dir_name (str) – Optional subdirectory name. By default logs are stored in ocs-ci-logs-<run_id>/<test_name>/failed_stress_job_logs directory. When dir_name is provided, logs are stored in ocs-ci-logs-<run_id>/<test_name>/failed_stress_job_logs/<dir_name>

ocs_ci.helpers.cephfs_stress_helpers.get_filtered_pods()

Gets a list of all pods running in the openshift-storage namespace, excluding specific patterns.

Returns:: list of filtered pod objects
Return type:: list

ocs_ci.helpers.cephfs_stress_helpers.get_mon_db_usage(): Retrieves the MON DB pod usage for all MON pods.

ocs_ci.helpers.cephfs_stress_helpers.get_mount_subdirs(pod_obj)

Retrieves a list of subdirectories located at the root of the PVC mount of the given pod.

Parameters:

pod_obj (obj) – pod object

Returns:

A list of directory names found at the mount path

Return type:

list

Raises:

ValueError – If mount path cannot be determined
CommandFailed – If command execution fails

ocs_ci.helpers.cephfs_stress_helpers.get_nodes_resource_utilization(): Gets the node cpu and memory utilization in percentage using ‘adm top’ and ‘oc describe’ for both master and worker node types

ocs_ci.helpers.cephfs_stress_helpers.get_osd_disk_utilization(): Gets disk utilization for individual OSDs and the total used capacity in the cluster.

ocs_ci.helpers.cephfs_stress_helpers.get_pods_resource_utilization(): Gets pod memory utilization using adm top command in raw output format.

ocs_ci.helpers.cephfs_stress_helpers.run_stress_cleanup(pod_obj, top_dir, timeout=3600, parallelism_count=25)

Executes a parallelized deletion of a directory structure.

This function utilizes ‘find’ combined with ‘xargs -P’ to spawn multiple deletion processes simultaneously.

Parameters:

pod_obj (obj) – The app pod obj to execute commands on
top_dir (str) – The relative directory name to delete (e.g: ‘cephfs-stress-job-xx’)
timeout (int, optional) – Max time in seconds to wait for cleanup. Defaults to 3600
parallelism_count (int, optional) – Number of concurrent ‘rm’ processes to spawn inside the pod. Defaults to 25

ocs_ci.helpers.cephfs_stress_helpers.verify_no_filesystem_hangs(stress_manager=None)

Verification function to check for filesystem hangs detected by monitoring.

Parameters:: stress_manager – CephFSStressTestManager instance to check pause status and get namespace
Returns:: True if no hangs detected, raises exception if hangs found
Return type:: bool
Raises:: Exception – If filesystem hangs are detected

ocs_ci.helpers.cephfs_stress_helpers.verify_openshift_storage_ns_pods_health(stress_manager=None)

Validates that all the Pods in the openshift-storage namespace are healthy. Retries on CommandFailed, then raises PodStabilityError if verification fails after all retries.

It checks for: 1. Pods with OOMKilled containers (fails the test) 2. Pods with restarts (informational only, logged as warning)

Parameters:: stress_manager – CephFSStressTestManager instance to check pause status
Raises:: PodStabilityError – If any pod is found to have OOMKilled containers after all retries

ocs_ci.helpers.cephfs_stress_helpers.verify_openshift_storage_ns_pods_in_running_state(stress_manager=None)

Verifies that all pods in the openshift-storage namespace are in a ‘Running’ state. Retries on CommandFailed, then raises PodsNotRunningError if verification fails after all retries.

Parameters:: stress_manager – CephFSStressTestManager instance to check pause status
Raises:: PodsNotRunningError – If not all pods are in the ‘Running’ state after all retries

ocs_ci.helpers.cluster_exp_helpers module

class ocs_ci.helpers.cluster_exp_helpers.BackgroundOps

Bases: object

EXPANSION_COMPLETED = False

wrap(func, *args, **kwargs)

Wraps the function to run specific iterations

Returns:: True if function runs successfully
Return type:: bool

class ocs_ci.helpers.cluster_exp_helpers.ClusterFiller(pods_to_fill, percent_required_filled, namespace)

Bases: object

Class for performing IOs on the pods

cluster_filler()

concurrent_copies = 5

filler(fill_pod)

This function copies the file downloaded by ‘downloader’ function in a unique directory to increase the cluster space utilization. Currently it makes 30 copies of the downloaded file in a given directory which is equivalent to almost 4 GiB of storage.

Parameters:: fill_pod – the pod on which the storage space need to be filled.

ocs_ci.helpers.cluster_exp_helpers.check_nodes_status(): This function runs in a loop to check the status of nodes. If the node(s) are in NotReady state then an exception is raised. Note: this function needs to be run as a background thread during the execution of a test

ocs_ci.helpers.cluster_exp_helpers.cluster_copy_ops(copy_pod)

Function to do copy operations in a given pod. Mainly used as a background IO during cluster expansion. It does of series of copy operations and verifies the data integrity of the files copied.

Parameters:: copy_pod (pod) – on which copy operations need to be done
Returns:: False, if there is data integrity check failure. Else, True
Return type:: Boolean

ocs_ci.helpers.cluster_exp_helpers.raw_block_io(raw_blk_pod, size='10G'): Runs the block ios on pod baased raw block pvc :param raw_blk_pod: pod on which block IOs should run :type raw_blk_pod: pod :param size: IO size :type size: str

ocs_ci.helpers.cnv_helpers module

Helper functions specific for CNV

ocs_ci.helpers.cnv_helpers.cal_md5sum_vm(vm_obj, file_path, username=None)

Calculate the MD5 checksum of a file via SSH on a virtual machine.

Parameters:

vm_obj (obj) – The virtual machine object.
file_path (str) – Full path to the file to calculate the MD5 checksum for.
username (str, optional) – The username to use for SSH authentication. Defaults to None.

Returns:

The MD5 checksum of the specified file.

Return type:

str

ocs_ci.helpers.cnv_helpers.calculate_vm_cnt_cpu_ram(cpu_per_vm=1, mem_per_vm=4, buffer=0.9)

Calculate the total and per-node virtual machine (VM) counts based on available CPU and memory resources.

This function determines the number of VMs that can be provisioned on each node in a Kubernetes cluster, considering both CPU and memory constraints. It uses the provided buffer factor to account for overheads.

Parameters:

cpu_per_vm (int) – The amount of CPU required per VM (default: 1).
mem_per_vm (int) – The amount of memory required per VM in GB (default: 4GB).
buffer (float) – The buffer percentage to reserve for node overheads (default: 0.9 or 90%).

Returns:

A tuple containing two dictionaries -: cluster_vm_count (dict): Total number of VMs that can be provisioned across the entire cluster. per_node_vm_count (dict): Number of VMs that can be provisioned on each node in the cluster.

Return type:

tuple

ocs_ci.helpers.cnv_helpers.check_fio_status(vm_obj, fio_service_name='fio_test'): Check if FIO is running after restart.

ocs_ci.helpers.cnv_helpers.clone_dv(source_pvc_name, source_pvc_ns, destination_ns)

Clones a DV using a specified data source

Parameters:

source_pvc_name (str) – PVC name of source vm used for cloning.
source_pvc_ns (str) – PVC namespace of source vm used for cloning.
destination_ns (str) – Namespace of cloned dv to be created on

Returns:

Cloned DV object

Return type:

dv_obj

ocs_ci.helpers.cnv_helpers.compute_vm_count_from_storage_capacity()

Computes the number of VMs to create based on available storage capacity.

This function calculates the number of VMs to create by determining the usable storage capacity after accounting for used storage and then dividing it by the VM size (30 GiB).

Returns:: The computed number of VMs to create.
Return type:: int

ocs_ci.helpers.cnv_helpers.convert_ssh_key_to_base64(ssh_key)

Convert SSH key to base64 encoding

Parameters:: ssh_key (str) – SSH key
Returns:: Base64 encoded SSH key
Return type:: str

ocs_ci.helpers.cnv_helpers.create_dv(access_mode='ReadWriteMany', sc_name='ocs-storagecluster-ceph-rbd-virtualization', pvc_size='30Gi', source_url='docker://quay.io/containerdisks/centos-stream:9', namespace='openshift-cnv')

Creates a DV using a specified data source

Parameters:

access_mode (str) – The access mode for the volume. Default is constants.ACCESS_MODE_RWX
sc_name (str) – The name of the storage class to use. Default is constants.DEFAULT_CNV_CEPH_RBD_SC.
pvc_size (str) – The size of the PVC. Default is “30Gi”.
source_url (str) – The URL of the vm registry image. Default is constants.CNV_CENTOS_SOURCE.
namespace (str, optional) – The namespace to create the DV on.

Returns:

DV object

Return type:

dv_obj

ocs_ci.helpers.cnv_helpers.create_fio_service(vm_obj, fio_cmd, fio_service_name)

Creates a systemd service on the given VM to run the specified FIO command persistently, ensuring it starts automatically after VM reboots.

Parameters:

vm_obj (str) – Name or reference to the virtual machine object.
fio_cmd (str) – The FIO command to be executed by the service.
fio_service_name (str) – Name of the systemd service to be created.

ocs_ci.helpers.cnv_helpers.create_pvc_using_data_source(source_name, access_mode='ReadWriteMany', sc_name='ocs-storagecluster-ceph-rbd-virtualization', pvc_size='30Gi', namespace='openshift-cnv')

Create a PVC using a specified data source

Parameters:

access_mode (str) – The access mode for the volume. Default is constants.ACCESS_MODE_RWX
source_name (str) – Name of the data source (VolumeImportSource) for the PVC
pvc_size (str) – Size of the PVC
sc_name (str) – StorageClass name for the PVC
namespace (str) – The namespace in which to create the PVC

Returns:

PVC object

Return type:

pvc_obj

ocs_ci.helpers.cnv_helpers.create_vm_secret(path=None, secret_name=None, namespace='openshift-cnv')

Create an SSH secret for the VM

Parameters:

path (str) – Path to the SSH public key file - optional
secret_name (str, optional) – Name of the secret. If not provided, a unique name will be generated.
namespace (str, optional) – Namespace in which the secret will be created. Defaults to constants.CNV_NAMESPACE.

Returns:

An OCS instance

Return type:

secret_obj

ocs_ci.helpers.cnv_helpers.create_volume_import_source(name=None, url='docker://quay.io/containerdisks/centos-stream:9')

Create a VolumeImportSource object

Parameters:

name (str) – Name for the VolumeImportSource. If not provided, a unique name will be generated
url (str) – URL for the registry source

Returns:

The created VolumeImportSource object

Return type:

source_data_obj

ocs_ci.helpers.cnv_helpers.expand_pvc_and_verify(vm_obj, new_size)

Expands the PVC for a VM and verifies the new size of pvc from inside the VM.

Parameters:

vm_obj – The VM object.
new_size (int) – The new PVC size in GB.

Returns:

True if expansion is successful.

Return type:

bool

Raises:

ValueError – If the pvc size is not expanded.

ocs_ci.helpers.cnv_helpers.generate_vm_password(length=10)

Generates a strong password for virtual machines with specified length. The password will contain at least one lowercase letter, one uppercase letter, one digit, and one punctuation symbol, with the rest filled randomly from all categories.

Parameters:: length (int) – The desired length of the password. Default is 10.
Returns:: A strong password meeting the criteria.
Return type:: str

ocs_ci.helpers.cnv_helpers.get_pvc_from_vm(vm_obj)

Get the PVC name from VM obj

Returns:: PVC in the form of ocs object
Return type:: ocs_ci.ocs.resources.ocs.OCS (obj)

ocs_ci.helpers.cnv_helpers.get_secret_from_vm(vm_obj)

Get the secret name from VM obj

Returns:: Secret in the form of ocs object
Return type:: ocs_ci.ocs.resources.ocs.OCS (obj)

ocs_ci.helpers.cnv_helpers.get_ssh_private_key_path()

Get the full path of the derived private key file from the associated SSH public key file

Returns:: The full path of the derived private key file
Return type:: str

ocs_ci.helpers.cnv_helpers.get_ssh_pub_key_with_filename(path=None)

Retrieve the content of the SSH public key and its file name

Parameters:: path (str) – Path to the SSH public key file - Optional
Returns:: A tuple containing the content of the SSH public key and the file name
Return type:: tuple

Raises: FileNotFoundError: If the provided ssh pub key path does not exist

ocs_ci.helpers.cnv_helpers.get_volumeimportsource(pvc_obj)

Get the volumeimportsource name from PVC obj

Returns:: volumeimportsource in the form of ocs object
Return type:: ocs_ci.ocs.resources.ocs.OCS (obj)

ocs_ci.helpers.cnv_helpers.install_fio_on_vm(vm_obj)

Detects the OS distribution of a virtual machine running in OpenShift Virtualization (CNV) and installs the ‘fio’ package using the appropriate package manager.

Parameters:: vm_obj (str) – Name of the virtual machine object.
Returns:: Output of the installation command.
Return type:: str

ocs_ci.helpers.cnv_helpers.run_dd_io(vm_obj, file_path, size='10240', username=None, verify=False)

Perform input/output (I/O) operation using dd command via SSH on a virtual machine.

Parameters:

vm_obj (obj) – The virtual machine object.
file_path (str) – The full path of the file to write on
size (str, optional) – Size in MB. Defaults to “102400” which is 10GB.
username (str, optional) – The username to use for SSH authentication. Defaults to None.
verify (bool, optional) – Whether to verify the I/O operation by calculating MD5 checksum. Defaults to False.

Returns:

If verify is True, returns the MD5 checksum of the written file. Otherwise, None.

Return type:

str or None

ocs_ci.helpers.cnv_helpers.run_fio(vm_obj, size='1G', io_direction='randrw', jobs=1, runtime=300, depth=4, rate='1m', bs='4K', direct=1, verify=True, verify_method='crc32c', filename='/testfile', fio_log_path='/tmp/fio_output.log', fio_service_name='fio_test')

Execute FIO on a CNV Virtual Machine with data integrity checks.

Parameters:

vm_obj – Name of the virtual machine object
size (str) – Size of the test file (e.g., ‘1G’).
io_direction (str) – Read/write mode (‘rw’, ‘randwrite’, ‘randread’).
jobs (int) – Number of FIO jobs to run.
runtime (int) – Duration of IO test (seconds).
depth (int) – I/O depth.
rate (str) – I/O rate limit.
bs (str) – Block size (default: ‘4K’).
direct (int) – Use direct I/O (1 = Yes, 0 = No).
verify (bool) – Enable data integrity verification.
verify_method (str) – Data integrity check method (‘crc32c’, ‘md5’, etc.).
filename (str) – Path of the test file in the VM.
fio_log_path (str) – Path where FIO logs will be stored.
fio_service_name (str) – name of fio service to be create

ocs_ci.helpers.cnv_helpers.verify_hotplug(vm_obj, disks_before_hotplug)

Verifies if a disk has been hot-plugged into/removed from a VM.

Parameters:

disks_before_hotplug (str) – Set of disk information before hot-plug or add.
vm_obj (VM object) – The virtual machine object to check.

Returns:

True if a hot-plugged disk is detected, False otherwise.

Return type:

bool

ocs_ci.helpers.cnv_helpers.verifyvolume(vm_name, volume_name, namespace)

Verify a volume in VM.

Parameters:

vm_name (str) – Name of the virtual machine
volume_name (str) – Name of the volume (PVC) to verify
namespace (str) – Virtual Machine Namespace

Returns:

True if the volume (PVC) is found, False otherwise

Return type:

bool

ocs_ci.helpers.disconnected module

ocs_ci.helpers.disconnected.get_oc_mirror_tool(): Download and install oc mirror tool.

ocs_ci.helpers.disconnected.get_opm_tool(): Download and install opm tool.

ocs_ci.helpers.disruption_helpers module

class ocs_ci.helpers.disruption_helpers.Disruptions

Bases: object

This class contains methods of disrupt operations

check_new_pid(node_name=None)

Check if the pid of the daemon has changed from the initially selected pid(daemon_pid attribute)

Parameters:: node_name (str) – Name of node in which the resource daemon is running

cluster_kubeconfig = ''

daemon_pid = None

delete_resource(resource_id=0)

kill_daemon(node_name=None, check_new_pid=True, kill_signal='9')

Kill self.resource daemon

Parameters:

node_name (str) – Name of node in which the resource daemon has to be killed
check_new_pid (bool) – True to check for new pid after killing the daemon. False to skip the check.
kill_signal (str) – kill signal type

kubeconfig_parameter()

Returns the ‘–kubeconfig <value>’ parameter for the oc command

Returns:

The ‘–kubeconfig <value>’ parameter for oc command if the attribute ‘cluster_kubeconfig’ is not empty.: Empty string if the the attribute ‘cluster_kubeconfig’ is empty.

Return type:

str

resource = None

resource_count = 0

resource_obj = None

select_daemon(node_name=None)

Select pid of self.resource daemon

Parameters:: node_name (str) – Name of node in which the resource daemon has to be selected.

selector = None

set_resource(resource, leader_type='provisioner', cluster_index=None)

class ocs_ci.helpers.disruption_helpers.FIOIntegrityChecker(pvc_factory, pod_factory, interfaces=None, pvc_size=5, max_latency_sec=None)

Bases: object

Verifies data integrity on RBD and CephFS PVCs during disruptive operations using a three-phase approach:

Write phase: Write a known file with FIO (size-based). Compute and store md5sum of each file.
Background IO phase: Start a time-based FIO (randrw) that runs throughout the disruptive operation. FIO results are written to a file inside the pod so they can be recovered if the oc rsh connection drops.
Verify phase: Wait for FIO to complete, check IO errors, p99 latency, and verify FIO runtime covered the operation. If oc rsh connection dropped during the operation, recover FIO results from the output file. Re-compute md5sum of the integrity file and compare with the stored checksum.

Usage:

checker = FIOIntegrityChecker(pvc_factory, pod_factory)
checker.start_io(bg_runtime=500)

# ... perform disruptive operation ...

checker.wait_and_verify()

For node removal operations where IO pods may be killed:

checker.verify_md5sum_only()

BG_IO_FILE = 'bg_io_data'

DEFAULT_MAX_LATENCY_SEC = 10

FIO_RESULTS_FILE = 'fio_results.json'

INTEGRITY_FILE = 'integrity_data'

start_io(size='1G', bg_runtime=900, bs='4K', rate='4m', node_name=None)

Create PVCs and pods, write integrity files with FIO, compute md5sums, then start background FIO (time-based, randrw).

Background FIO writes results to a file inside the pod (fio_results.json) so they can be recovered if the oc rsh connection drops during the disruptive operation.

Parameters:

size (str) – Size of the integrity file to write.
bg_runtime (int) – Background FIO runtime in seconds. Should be longer than the expected operation duration to ensure IO coverage.
bs (str) – Block size for background FIO.
rate (str) – IO rate limit for background FIO.
node_name (str) – Pin IO pods to this specific node. Useful when other nodes may be removed during the operation.

verify_md5sum_only()

Re-compute md5sum of the integrity files and compare with the stored checksums. Does not check FIO results.

Use this after a disruptive operation where IO pods may have been killed (e.g. node removal that removed the pod’s node).

Raises:: AssertionError – If md5sum does not match or no pods were reachable for verification.

wait_and_verify()

Wait for background FIO to complete, validate results, and verify data integrity.

Checks performed: - FIO IO error counters (read/write errors == 0) - p99 completion latency under max_latency_sec threshold - FIO runtime covered the disruptive operation duration - md5sum of integrity file matches pre-operation checksum

If oc rsh connection dropped during the operation, FIO results are recovered from the output file inside the pod.

Raises:

AssertionError – If FIO reported IO errors, latency exceeded threshold, or md5sum does not match.
CommandFailed – If FIO results cannot be recovered.

ocs_ci.helpers.disruption_helpers.delete_resource_multiple_times(resource_name, num_of_iterations)

Delete a specific resource(osd, rook-operator, mon, etc,.) multiple times.

Parameters:

resource_name (str) – The resource name to delete
num_of_iterations (int) – The number of iterations we delete the resource

ocs_ci.helpers.dr_helpers module

Helper functions specific for DR

ocs_ci.helpers.dr_helpers.add_label_to_appsub(workloads, label='test', value='test1')

Function to add new label with any value to the AppSub on the hub. This is needed as WA for sub app pods to show up after failover in ACM 2.11 post hub recovery (bz: 2295782)

Parameters:

workloads (list) – List of workloads created
label (str) – Name of label to be added
value (str) – Value to be added

ocs_ci.helpers.dr_helpers.apply_drpolicy_to_workload(workload, drcluster_name)

Function for applying drpolicy to indiviusual workload

Parameters:

workload (List) – List of workload objects
drcluster_name (str) – Name of the DRcluster on which workloads belongs

ocs_ci.helpers.dr_helpers.apply_itms_to_managed_clusters(itms_file_path)

Apply itms configuration to all managed clusters and wait for MCP to complete.

Parameters:: itms_file_path (str) – Path to the itms-oc-mirror.yaml file

ocs_ci.helpers.dr_helpers.check_mirroring_status_for_custom_pool(pool_name, namespace='openshift-storage', min_replaying=1)

Check the health and mirroring status of a custom CephBlockPoolRadosNamespace resource. Refer For OCSQE-2294 or RHSTOR-5129 in ODF 4.19 for details

This function verifies that: - At least two such resources exist in the given namespace - The specified pool has all health fields set to ‘OK’ - The replaying count in both ‘image_states’ and ‘states’ meets the minimum threshold

Parameters:

pool_name (str) – Base name of the Ceph block pool (without ‘-builtin-implicit’ suffix) whose
validated. (mirroring status has to be) –
namespace (str) – Namespace to look for the resource. Default is ‘openshift-storage’.
min_replaying (int) – Minimum expected value for replaying count. Default is 1.

Returns:

True if all checks pass, otherwise False.

Return type:

bool

Raises:

ValueError – If custom Pool is missing, insufficient Pool count, or summary is not found.

ocs_ci.helpers.dr_helpers.check_mirroring_status_ok(replaying_images=None, replaying_groups=None, cephblockpoolradosns=None, storageclient_uid=None)

Check if mirroring status has health OK and expected number of replaying images and groups.

Parameters:

replaying_images (int) – Expected number of images in replaying state
replaying_groups (int) – Expected number of groups in replaying state. Applicable when CG is enabled.
cephblockpoolradosns (string) – The name of the cephblockpoolradosnamespace
storageclient_uid (string) – The uid of the storageclient in the client cluster where the application is running. Applicable for provider - client configuration.

Returns:

True if status contains expected health and states values, False otherwise

Return type:

bool

Raises:

NotFoundError – If the configuration is provider mode and the name of the cephblockpoolradosnamespace is not obtained

ocs_ci.helpers.dr_helpers.check_rbd_mirror_running(namespace=None)

Check if the rbd-mirror daemon deployment is running with at least one ready replica. Ceph HEALTH_OK does not reflect rbd-mirror daemon absence, so this explicit check is needed to catch silent failures before tests run.

Parameters:: namespace (str) – Namespace to check in. Defaults to config.ENV_DATA[‘cluster_namespace’].
Returns:: True if rbd-mirror deployment has ready replicas
Return type:: bool
Raises:: UnexpectedDeploymentConfiguration – If the rbd-mirror deployment is not found or not running

ocs_ci.helpers.dr_helpers.check_replication_resource_state(kind, state, namespace, resource_name='')

Check if replication resources in the given namespace are in expected state

Parameters:

kind (str) – Kind of resource (e.g., constants.VOLUME_REPLICATION, constants.VOLUME_REPLICATION_GROUP, etc.)
state (str) – The resource state to check for (e.g. ‘primary’, ‘secondary’)
namespace (str) – the namespace of the resources
resource_name (str) – Name of specific resource

Returns:

True if resources are in expected state or were deleted, False otherwise

Return type:

bool

ocs_ci.helpers.dr_helpers.check_resource_existence(kind, namespace, resource_name='')

Check if resource exists in the given namespace

Parameters:

kind (str) – Kind of resource (e.g., constants.VOLUME_REPLICATION_GROUP, constants.VOLUME_GROUP_REPLICATION)
namespace (str) – the namespace of the resource
resource_name (str) – Name of resource

Returns:

True if resource exists, False otherwise

Return type:

bool

ocs_ci.helpers.dr_helpers.check_storage_cluster_peer_state()

Checks Storage cluster peer state

Returns:: True if storage cluster peer state is ‘Peered’. otherwise False
Return type:: bool

ocs_ci.helpers.dr_helpers.configure_drcluster_for_fencing(): Configures DRClusters for enabling fencing

ocs_ci.helpers.dr_helpers.configure_rdr_hub_recovery()

RDR helper function to create backup schedule on the active hub cluster needed for hub recovery using backup and restore.

This function ensures all pre-reqs are verified before hub recovery is performed.

ocs_ci.helpers.dr_helpers.create_backup_schedule(): Create backupschedule resource only on active hub

ocs_ci.helpers.dr_helpers.create_ingress_cert_dr(cert_name='user-ca-bundle', namespace='openshift-config', patch_proxy=True)

ocs_ci.helpers.dr_helpers.create_klusterlet_config(): Create klusterletconfig after hub recovery to avoid eviction of resources by adding “AppliedManifestWork” eviction grace period

ocs_ci.helpers.dr_helpers.create_multiclusterservice_dr()

This function is used to create multiClusterService used for RDR

Returns:: true when multiClusterService already exists
Return type:: bool

ocs_ci.helpers.dr_helpers.create_service_exporter(annotate=True)

Create Service exporter

Parameters:: annotate (bool) – If True - annotate the service exporter

ocs_ci.helpers.dr_helpers.disable_dr_from_app(secondary_cluster_name)

Function to disable DR from app

Parameters:: secondary_cluster_name (str) – cluster where application is running

ocs_ci.helpers.dr_helpers.disable_dr_rdr(discovered_apps=False): Disable DR for the applications

ocs_ci.helpers.dr_helpers.do_discovered_apps_cleanup(drpc_name, old_primary, workload_namespace, workload_dir, vrg_name, skip_resource_deletion_verification=False, ignore_resource_not_found=False)

Function to clean up Resources

Parameters:

drpc_name (str) – Name of DRPC
old_primary (str) – Name of old primary where cleanup will happen
workload_namespace (str) – Workload namespace
workload_dir (str) – Dir location of workload
vrg_name (str) – Name of VRG
skip_resource_deletion_verification (bool) – False by default and runs always, else resource verification is handled separately in the test when Shared protection type is used for DR protection via ACM UI
ignore_resource_not_found (bool) – False by default, resource not found is ignored when the workload which was DR protected via ACM UI is deleted, refer DFBUGS-3706

ocs_ci.helpers.dr_helpers.do_discovered_apps_cleanup_multi_ns(old_primary, workload_instance, vrg_state='secondary')

Function to clean up Resources

Parameters:

old_primary (str) – Name of old primary where cleanup will happen
workload_instance (list) – Workload instance

ocs_ci.helpers.dr_helpers.enable_fence(drcluster_name, switch_ctx=None)

Once the managed cluster is fenced, all communication from applications to the ODF external storage cluster will fail

Parameters:

drcluster_name (str) – Name of the DRcluster which needs to be fenced
switch_ctx (int) – The cluster index by the cluster name

ocs_ci.helpers.dr_helpers.enable_unfence(drcluster_name, switch_ctx=None)

The OpenShift cluster to be Unfenced is the one where applications are not currently running and the cluster that was Fenced earlier.

Parameters:

drcluster_name (str) – Name of the DRcluster which needs to be fenced
switch_ctx (int) – The cluster index by the cluster name

ocs_ci.helpers.dr_helpers.extract_images_from_yaml(obj, images=None)

Recursively extract container image references from a YAML object. Extracts values from ‘image’, ‘url’, and ‘value’ keys that contain image references.

Parameters:

obj – YAML object (dict, list, or primitive)
images – Set to collect images (created if None)

Returns:

Set of container image strings

Return type:

set

ocs_ci.helpers.dr_helpers.failover(failover_cluster, namespace, workload_type='Subscription', workload_placement_name=None, switch_ctx=None, discovered_apps=False, old_primary=None)

Initiates Failover action to the specified cluster

Parameters:

failover_cluster (str) – Cluster name to which the workload should be failed over
namespace (str) – Namespace where workload is running
workload_type (str) – Type of workload, i.e., Subscription or ApplicationSet
workload_placement_name (str) – Placement name
switch_ctx (int) – The cluster index by the cluster name
discovered_apps (bool) – True when cluster is failing over DiscoveredApps
old_primary (str) – Name of cluster where workload were running

ocs_ci.helpers.dr_helpers.fence_state(drcluster_name, fence_state, switch_ctx=None)

Sets the specified clusterFence state

Parameters:

drcluster_name (str) – Name of the DRcluster which needs to be fenced
fence_state (str) – Specify the clusterfence state either constants.ACTION_UNFENCE and ACTION_FENCE
switch_ctx (int) – The cluster index by the cluster name

ocs_ci.helpers.dr_helpers.generate_kubeobject_capture_interval()

Generate KubeObject Capture Interval

Returns:: capture interval value to be used
Return type:: int

ocs_ci.helpers.dr_helpers.generate_rdr_mirror_images()

Extract and return list of container images from RDR workload repository.

Returns:: List of container image strings for mirroring in disconnected environments
Return type:: list

ocs_ci.helpers.dr_helpers.get_all_drclusters()

Get all DRClusters

Returns:: List of all DRClusters
Return type:: list

ocs_ci.helpers.dr_helpers.get_all_drpolicy()

Gets all DRPolicy from hub cluster

Returns:: List of all DRPolicy
Return type:: list

ocs_ci.helpers.dr_helpers.get_backend_volumes_for_pvcs(namespace)

Gets list of RBD images or CephFS subvolumes associated with the PVCs in the given namespace

Parameters:: namespace (str) – The namespace of the PVC resources
Returns:: List of RBD images or CephFS subvolumes
Return type:: list

ocs_ci.helpers.dr_helpers.get_cluster_set_name(switch_ctx=None)

Get Cluster set name from managedcluster

Returns:: List of uniq cluster set name
Return type:: list

ocs_ci.helpers.dr_helpers.get_current_primary_cluster_name(namespace, workload_type='Subscription', discovered_apps=False, resource_name=None)

Get current primary cluster name based on workload namespace

Parameters:

namespace (str) – Name of the namespace
workload_type (str) – Type of workload, i.e., Subscription or ApplicationSet
discovered_apps (bool) – If true then deployed workload is discovered_apps
resource_name (str) – DRPC NAME Only Used for discovered apps

Returns:

Current primary cluster name

Return type:

str

ocs_ci.helpers.dr_helpers.get_current_secondary_cluster_name(namespace, workload_type='Subscription', discovered_apps=False, resource_name=None)

Get current secondary cluster name based on workload namespace

Parameters:

namespace (str) – Name of the namespace
workload_type (str) – Type of workload, i.e., Subscription or ApplicationSet
discovered_apps (bool) – If true then deployed workload is discovered_apps
resource_name (str) – DRPC NAME Only Used for discovered apps

Returns:

Current secondary cluster name

Return type:

str

ocs_ci.helpers.dr_helpers.get_fence_state(drcluster_name, switch_ctx=None)

Returns the clusterfence state of given drcluster

Parameters:

drcluster_name (str) – Name of the DRcluster
switch_ctx (int) – The cluster index by the cluster name

Returns:

If drcluster are fenced: Fenced or Unfenced, else None if not defined

Return type:

state (str)

ocs_ci.helpers.dr_helpers.get_fencing_cidrs_from_drclusterconfig(cluster_name)

Read fencing CIDRs from DRClusterConfig.status.storageAccessDetails on the current (managed) cluster context (ODF 4.21+ / Ramen).

Prefers the DRClusterConfig named like the managed cluster, then RBD CSI provisioner entries, with sensible fallbacks.

Parameters:: cluster_name (str) – Managed cluster name (matches DRCluster / DRClusterConfig name on hub)
Returns:: CIDR strings for hub DRCluster.spec.cidrs
Return type:: list
Raises:: UnexpectedBehaviour – If CIDRs are not yet published or cannot be determined

ocs_ci.helpers.dr_helpers.get_managed_cluster_node_ips()

Gets node ips of individual managed clusters for enabling fencing from each managed cluster’s DRClusterConfig

Returns:: [[managed_cluster_name, multicluster_index, [cidr, …]], …]
Return type:: list

ocs_ci.helpers.dr_helpers.get_nodes_from_active_zone(namespace)

Get the nodes list and index from active zone

Parameters:

namespace (str) – Namespace of the app workload

Returns:

contains index and the node_objs list of the cluster: active_hub_index (int): Index of the active hub cluster active_hub_cluster_node_objs (list): Node list of the active hub nodes managed_cluster_index (int): Index of the active zone managed cluster managed_cluster_node_objs (list): Node list of the active zone managed cluster ceph_node_ips (list): Ceph node list which are running in active zone

Return type:

tuple

ocs_ci.helpers.dr_helpers.get_resource_count(kind, namespace=None)

Gets resource count in given namespace for specified resource kind

Parameters:

kind (str) – Kind of resource (e.g., constants.VOLUME_REPLICATION, constants.REPLICATION_SOURCE, etc.)
namespace (str) – the namespace of the resources

Returns:

Resource count

Return type:

int

ocs_ci.helpers.dr_helpers.get_scheduling_interval(namespace, workload_type='Subscription', discovered_apps=False, resource_name=None)

Get scheduling interval for the workload in the given namespace

Parameters:

namespace (str) – Name of the namespace
workload_type (str) – Type of workload, i.e., Subscription or ApplicationSet
discovered_apps (bool) – If true then deployed workload is discovered_apps

Returns:

scheduling interval value from DRPolicy

Return type:

int

ocs_ci.helpers.dr_helpers.get_vgs_name(vgs_namespace)

Fetches the name of Volume Group Snapshot from Replication Group Source

Parameters:: namespace (str) – the namespace of the Volume Group Snapshot

ocs_ci.helpers.dr_helpers.gracefully_reboot_ocp_nodes(drcluster_name, disable_eviction=False)

Gracefully reboot OpenShift Container Platform nodes which was fenced before

Parameters:

drcluster_name (str) – Name of the drcluster which needs to be rebooted
disable_eviction (bool) – On True will delete pod that is protected by PDB, False by default

ocs_ci.helpers.dr_helpers.is_cg_cephfs_enabled()

Validate if consistency group is enabled

Returns: True, if volume group snapshot class exists. False otherwise

ocs_ci.helpers.dr_helpers.is_cg_enabled()

Check if Consistency Group feature is enabled via environment variable

Returns:: True if CG is enabled, False otherwise
Return type:: bool

Note

CG can be enabled/disabled via cg_enabled in dr_workload.yaml config.

ocs_ci.helpers.dr_helpers.mdr_post_failover_check(namespace, timeout=1200)

Post the failover verify that Pod is been deleted from primary cluster and PVCs are in terminating state

Parameters:

namespace (str) – Namespace of the application
timeout (int) – time in seconds to wait for resource deletion

ocs_ci.helpers.dr_helpers.ordered_unique_cidrs(cidrs): Preserve order while removing duplicates

ocs_ci.helpers.dr_helpers.relocate(preferred_cluster, namespace, workload_type='Subscription', workload_placement_name=None, switch_ctx=None, discovered_apps=False, old_primary=None, workload_instance=None, multi_ns=False, workload_instances_shared=None, vm_auto_cleanup=False)

Initiates Relocate action to the specified cluster

Parameters:

preferred_cluster (str) – Cluster name to which the workload should be relocated
namespace (str) – Namespace where workload is running
workload_type (str) – Type of workload, i.e., Subscription or ApplicationSet
workload_placement_name (str) – Placement name
switch_ctx (int) – The cluster index by the cluster name
discovered_apps (bool) – If true then deployed workload is discovered_apps
old_primary (str) – Name of cluster where workload were running
workload_instance (object) – Discovered App instance to get namespace and dir location
multi_ns (bool) – Multi Namespace
workload_instances_shared (list) – List of workloads tied to a single DRPC using Shared Protection type
vm_auto_cleanup (bool) – If true, cleanup will not be initiated after relocate action, False otherwise.

ocs_ci.helpers.dr_helpers.remove_parameter_klusterlet_config(): Edit the global KlusterletConfig on the new hub and remove the parameter appliedManifestWorkEvictionGracePeriod and its value

ocs_ci.helpers.dr_helpers.replace_cluster(workload, primary_cluster_name, secondary_cluster_name)

Function to do core replace cluster task

Parameters:

workload (List) – List of workload objects
primary_cluster_name (str) – Name of the primary DRcluster
secondary_cluster_name (str) – Name of the secondary DRcluster

ocs_ci.helpers.dr_helpers.restore_backup(): Restores the backup in new hub and make it as active

ocs_ci.helpers.dr_helpers.set_current_primary_cluster_context(namespace, workload_type='Subscription')

Set current primary cluster context based on workload namespace

Parameters:

namespace (str) – Name of the namespace
workload_type (str) – Type of workload, i.e., Subscription or ApplicationSet

ocs_ci.helpers.dr_helpers.set_current_secondary_cluster_context(namespace, workload_type='Subscription')

Set secondary cluster context based on workload namespace

Parameters:

namespace (str) – Name of the namespace
workload_type (str) – Type of workload, i.e., Subscription or ApplicationSet

ocs_ci.helpers.dr_helpers.setup_fdf_catsrc_for_hub(): This function creates fdf catalogsource on hub

ocs_ci.helpers.dr_helpers.validate_drpolicy_grouping(drpolicy_name=None)

Validate DRPolicy configuration for CG behavior.

This function validates that DRPolicy has grouping=true for every storageClass in status.async.peerClasses in ODF version >= 4.21

Parameters:: drpolicy_name (str, optional) – Name of specific DRPolicy to validate. If None, validates all DRPolicies.
Returns:: True if DRPolicy grouping validation passes
Return type:: bool
Raises:: UnexpectedBehaviour – If peerClasses are not found or grouping validation fails

ocs_ci.helpers.dr_helpers.validate_protection_label(kind, namespace, protection_name=None)

Gets the yaml file for specified resource kind in the given namespace

Parameters:

kind (str) – Kind of resource (e.g., constants.VM, constants.PVC, etc.)
namespace (str) – the namespace of the specified resource
protection_name (str) – name of protection in UI

Raises:

AssertionError – If the protection label is not found on any of the resources of the
specified kind in the given namespace –

ocs_ci.helpers.dr_helpers.validate_storage_cluster_peer_state()

Validate Storage cluster peer state

Raises:: TimeoutExpiredError – incase storage cluster peer state is not reached ‘Peered’ state.

ocs_ci.helpers.dr_helpers.validate_vgrc_count()

Validate VGRC count on each managed cluster per unique scheduling interval.

This function collects all unique scheduling intervals from all DRPolicies and validates that the VGRC count matches the number of unique intervals.

Returns:: True if VGRC count validation passes, False otherwise
Return type:: bool
Raises:: AssertionError – If VGRC count validation fails

ocs_ci.helpers.dr_helpers.validate_volumegroupsnapshot(vgs_namespace)

Validates Volume Group Snapshot resource creation from odf external snapshotter

Parameters:: vgs_namespace (str) – the namespace of the Volume Group snapshot resources

ocs_ci.helpers.dr_helpers.verify_backend_volume_deletion(backend_volumes, cephblockpoolradosns=None, cephfssubvolumegroup=None, storageclient_uid=None)

Check whether RBD images/CephFS subvolumes are deleted in the backend.

Parameters:

backend_volumes (list) – List of RBD images or CephFS subvolumes
cephblockpoolradosns (str) – The name of the cephblockpoolradosnamespace
cephfssubvolumegroup (str) – The name of the cephfilesystemsubvolumegroup
storageclient_uid (string) – The uid of the storageclient in the client cluster where the application is running. Applicable for provider - client configuration.

Returns:

True if volumes are deleted and False if volumes are not deleted

Return type:

bool

Raises:

NotFoundError – If the configuration is provider mode and the name of the cephblockpoolradosnamespace is not obtained

ocs_ci.helpers.dr_helpers.verify_backup_is_taken(): Function to verify backup is taken

ocs_ci.helpers.dr_helpers.verify_cluster_data_protected_status(workload_type, namespace, workload_placement_name=None)

Verify that the cluster dataProtected is True

Parameters:

workload_type (str) – Type of workload, i.e., Subscription or ApplicationSet
namespace (str) – the namespace of the drpc resources
workload_placement_name (str) – Placement name

ocs_ci.helpers.dr_helpers.verify_drcluster_validated_on_hub(drcluster_name, switch_ctx=None)

Wait until hub DRCluster reports a successful validation condition.

Ramen surfaces hub reconciliation via status.conditions (reason/type Validated or legacy Succeeded).

ocs_ci.helpers.dr_helpers.verify_drpc_placement_deletion(cmd, expected_output_lst)

Function to validate drpc deletion

Parameters:

cmd (str) – cli command
expected_output_lst (set) – A set of strings that need to be included in the command output.

Returns:

True, if all strings are included in the command output, False otherwise.

Return type:

bool

ocs_ci.helpers.dr_helpers.verify_drpolicy_cli(switch_ctx=None)

Function to verify DRPolicy status

Returns:: True if the status is in succeed state, else raise exception switch_ctx (int): The cluster index by the cluster name
Return type:: bool

ocs_ci.helpers.dr_helpers.verify_fence_state(drcluster_name, state, switch_ctx=None)

Verify the specified drcluster is in expected state

Parameters:

drcluster_name (str) – Name of the DRcluster
state (str) – The fence state it is either constants.ACTION_FENCE or constants.ACTION_UNFENCE
switch_ctx (int) – The cluster index by the cluster name

Returns:

True (bool) if the drcluster is in expected state

Raises:

Raises exception Unexpected-behaviour if the specified drcluster is not in the given state condition –

ocs_ci.helpers.dr_helpers.verify_last_group_sync_time(drpc_obj, scheduling_interval, initial_last_group_sync_time=None)

Verifies that the lastGroupSyncTime for a given DRPC object is within the expected range.

Parameters:

drpc_obj (obj) – DRPC object
scheduling_interval (int) – The scheduling interval in minutes
initial_last_group_sync_time (str) – Previous lastGroupSyncTime value (optional).

Returns:

Current lastGroupSyncTime

Return type:

str

Raises:

AssertionError – If the lastGroupSyncTime is outside the expected range (greater than or equal to three times the scheduling interval)

ocs_ci.helpers.dr_helpers.verify_last_kubeobject_protection_time(drpc_obj, kubeobject_sync_interval)

Verifies that the lastKubeObjectProtectionTime for a given DRPC object is within the expected range.

Parameters:

drpc_obj (obj) – DRPC object
kubeobject_sync_interval (int) – The KubeObject sync interval in minutes

Returns:

Current lastKubeObjectProtectionTime

Return type:

str

Raises:

AssertionError – If the lastKubeObjectProtectionTime is outside the expected range (greater than or equal to two times the scheduling interval)

ocs_ci.helpers.dr_helpers.verify_restore_is_completed(): Function to verify restore is completed or finished

ocs_ci.helpers.dr_helpers.verify_volsync(): Verify volsync pod is created in volsync-system namespace

ocs_ci.helpers.dr_helpers.wait_for_all_resources_creation(pvc_count, pod_count, namespace, timeout=900, skip_replication_resources=False, discovered_apps=False, vrg_name='', skip_vrg_check=False, performed_dr_action=False)

Wait for workload and replication resources to be created

Parameters:

pvc_count (int) – Expected number of PVCs
pod_count (int) – Expected number of Pods
namespace (str) – the namespace of the workload
timeout (int) – time in seconds to wait for resource creation
skip_replication_resources (bool) – if true vr status wont’t be check
discovered_apps (bool) – If true then deployed workload is discovered_apps
vrg_name (str) – Name of VRG
skip_vrg_check (bool) – If true vrg check will be skipped
performed_dr_action (bool) – It true, VolumeSnapshot count will be validate for CG-CephFS

ocs_ci.helpers.dr_helpers.wait_for_all_resources_deletion(namespace, timeout=1000, discovered_apps=False, workload_cleanup=False, vrg_name='', skip_vrg_check=False)

Wait for workload and replication resources to be deleted

Parameters:

namespace (str) – the namespace of the workload
timeout (int) – time in seconds to wait for resource deletion
discovered_apps (bool) – If true then deployed workload is discovered_apps
workload_cleanup (bool) – Set to True when performing final workload cleanup. If True: - PVC and PV deletion will always be checked - Replication resources state check will be skipped.
vrg_name (str) – Name of VRG
skip_vrg_check (bool) – If true vrg check will be skipped

ocs_ci.helpers.dr_helpers.wait_for_backend_volume_deletion(backend_volumes, timeout=600)

Verify that RBD image/CephFS subvolume are deleted in the backend.

Parameters:

backend_volumes (list) – List of RBD images or CephFS subvolumes
timeout (int) – time in seconds to wait

Raises:

TimeoutExpiredError – In case backend volumes are not deleted

ocs_ci.helpers.dr_helpers.wait_for_cnv_workload(vm_name, namespace, phase='Running', timeout=600)

Wait for VM to reach a phase

Parameters:

vm_name (str) – Name of the VM
namespace (str) – Namespace of the vm workload
phase (str) – Phase of the vm resource to wait for. example: Running, Stopped
timeout (int) – time in seconds to wait for resource deletion

ocs_ci.helpers.dr_helpers.wait_for_mirroring_status_ok(replaying_images=None, replaying_groups=None, timeout=900)

Wait for mirroring status to reach health OK and expected number of replaying images and groups for each of the ODF cluster.

Parameters:

replaying_images (int) – Expected number of images in replaying state
replaying_groups (int) – Expected number of groups in replaying state. Applicable when CG is enabled.
timeout (int) – time in seconds to wait for mirroring status reach OK

Returns:

True if status contains expected health and states values

Return type:

bool

Raises:

TimeoutExpiredError – In case of unexpected mirroring status

ocs_ci.helpers.dr_helpers.wait_for_replication_destinations_creation(rep_dest_count, namespace, timeout=900)

Wait for ReplicationDestination resources to be created

Parameters:

rep_dest_count (int) – Expected number of ReplicationDestination resource
namespace (str) – The namespace of the ReplicationDestination resources
timeout (int) – Time in seconds to wait for ReplicationDestination resources to be created

Raises:

TimeoutExpiredError – If expected number of ReplicationDestination resources not created

ocs_ci.helpers.dr_helpers.wait_for_replication_destinations_deletion(namespace, timeout=900)

Wait for ReplicationDestination resources to be deleted

Parameters:

namespace (str) – The namespace of the ReplicationDestination resources
timeout (int) – Time in seconds to wait for ReplicationDestination resources to be deleted

Raises:

TimeoutExpiredError – If expected number of ReplicationDestination resources not deleted

ocs_ci.helpers.dr_helpers.wait_for_replication_resources_creation(count, namespace, timeout, discovered_apps=False, vrg_name='', skip_vrg_check=False, performed_dr_action=False)

Wait for replication resources to be created

Parameters:

count (int) – Expected number of VR resources or ReplicationSource count
namespace (str) – the namespace of the VR or ReplicationSource resources
timeout (int) – time in seconds to wait for VR or ReplicationSource resources to be created or reach expected state
discovered_apps (bool) – If true then deployed workload is discovered_apps
vrg_name (str) – Name of VRG
skip_vrg_check (bool) – If true vrg check will be skipped

Raises:

TimeoutExpiredError – In case replication resources not created

ocs_ci.helpers.dr_helpers.wait_for_replication_resources_deletion(namespace, timeout, check_state=True, discovered_apps=False, vrg_name='', skip_vrg_check=False)

Wait for replication resources to be deleted

Parameters:

namespace (str) – the namespace of the resources’
timeout (int) – time in seconds to wait for resources to reach expected state or deleted
check_state (bool) – True for checking resources state before deletion, False otherwise
discovered_apps (bool) – If true then deployed workload is discovered_apps
vrg_name (str) – Name of VRG
skip_vrg_check (bool) – If true vrg check will be skipped

Raises:

TimeoutExpiredError – In case replication resources not deleted

ocs_ci.helpers.dr_helpers.wait_for_resource_count(kind, namespace, expected_count=1, timeout=900)

Wait for resources to reach expected count

Parameters:

kind (str) – Kind of resource (e.g., constants.VOLUME_REPLICATION, constants.REPLICATION_SOURCE, etc.)
namespace (str) – The namespace of the resources
expected_count (int) – Expected number of resources
timeout (int) – Time in seconds to wait

Raises:

TimeoutExpiredError – If expected number of resources not reached

ocs_ci.helpers.dr_helpers.wait_for_resource_existence(kind, namespace, resource_name='', timeout=900, should_exist=True)

Wait for resources to exist or not exist

Parameters:

kind (str) – Kind of resource (e.g., constants.VOLUME_REPLICATION_GROUP, constants.VOLUME_GROUP_REPLICATION)
namespace (str) – the namespace of the resource
resource_name (str) – Name of resource
timeout (int) – Time in seconds to wait
should_exist (bool) – True to wait for existence, False to wait for deletion

Raises:

TimeoutExpiredError – If expected resource state not reached

ocs_ci.helpers.dr_helpers.wait_for_resource_state(kind, state, namespace, resource_name='', timeout=900)

Wait for resources to reach expected count

Parameters:

kind (str) – Kind of resource (e.g., constants.VOLUME_REPLICATION, constants.VOLUME_REPLICATION_GROUP, etc.)
state (str) – The resource state to check for (e.g. ‘primary’, ‘secondary’)
namespace (str) – the namespace of the resources
resource_name (str) – Name of specific resource
timeout (int) – Time in seconds to wait

Raises:

TimeoutExpiredError – If expected number of resources not reached

ocs_ci.helpers.dr_helpers.wait_for_vrg_state(vrg_state, vrg_namespace, resource_name, timeout=900)

Wait for VRG state

Parameters:

vrg_state (str) – VRG expected state
vrg_namespace (str) – VRG resource namespace
resource_name (str) – VRG resource name
timeout (int) – Timeout for wait

ocs_ci.helpers.dr_helpers_ui module

Helper functions specific to DR User Interface

ocs_ci.helpers.dr_helpers_ui.application_count_on_ui(acm_obj)

The function fetches the total application count on the DR monitoring dashboard

Parameters:: acm_obj (AcmAddClusters) – ACM Page Navigator Class
Returns:: Number of ACM managed applications and total applications enrolled in disaster recovery on DR dashboard
Return type:: app_count_list (list)

ocs_ci.helpers.dr_helpers_ui.check_apps_running_on_selected_cluster(acm_obj, cluster_name, app_names: List[str], timeout=10)

Function to check the apps running on selected managed cluster on DR monitoring dashboard

Parameters:

acm_obj (AcmAddClusters) – ACM Page Navigator Class
cluster_name (str) – Name of the managed cluster where apps are primary
app_names (list) – Name of the multiple apps from CLI in the form of a list to iterate over it
timeout (int) – Timeout for which an element on UI should be checked for

Returns:

True if all the apps are found on selected managed cluster, False if any of the apps are missing

ocs_ci.helpers.dr_helpers_ui.check_cluster_operator_status(acm_obj, timeout=30)

The function verifies the cluster operator status on the DR monitoring dashboard

Parameters:

acm_obj (AcmAddClusters) – ACM Page Navigator Class
timeout (int) – Timeout for which status check should be done

Returns:

False if expected text Degraded is found, True otherwise

Return type:

bool

ocs_ci.helpers.dr_helpers_ui.check_cluster_status_on_acm_console(acm_obj, down_cluster_name=None, cluster_names=None, timeout=900, expected_text='Ready')

This function checks the current status of imported clusters on the ACM console. These clusters are the managed OCP clusters and the ACM Hub cluster.

Parameters:

acm_obj (AcmAddClusters) – ACM Page Navigator Class
down_cluster_name (str) – If Failover is performed when a cluster goes down, it waits and checks the updated status of cluster unavailability on the ACM console. It takes the cluster name which is down.
cluster_names (list) – This is a list of cluster names involved in a DR setup. You can either pass the cluster names as args in the form of list, but if not passed, it fetches the primary & secondary cluster names passed at run time for context setting (max. 3 for now including ACM Hub cluster). ACM Hub cluster name is hard coded as “local-cluster” as the name is constant & isn’t expected to change.
timeout (int) – Timeout to wait for certain elements to be found on the ACM UI
expected_text (str) – Any particular string/status of the cluster to be checked on the ACM console. Default is set to ready

ocs_ci.helpers.dr_helpers_ui.check_or_assign_drpolicy_for_discovered_vms_via_ui(acm_obj, vms: List[object], managed_cluster_name, assign_policy=True, standalone=True, protection_name=None, namespace=None)

This function can be used to check the VM status and assign Data Policy using UI to Discovered VMs via Virtual machines page of the ACM console. Starting ODF 4.19 and ACM 2.14, Data Policy can be assigned as Standalone or Shared Protection type (if there is an existing DRPC for another VM workload, and you want to club it together)

Parameters:

acm_obj (AcmAddClusters) – ACM Page Navigator Class
vms (object) – Contains object of VMs for DR protection in the form of a list
managed_cluster_name (str) – Name of the managed cluster where VM workload is running
assign_policy (bool) – Optional steps when only VM status has to be checked, DRPolicy won’t be applied when False
standalone (bool) – True by default, will switch to Shared Protection type when False
protection_name (str) – Protection name used to DR protect the workload using which DRPC and Placement would be created
namespace – None by default, namespace of the workload

ocs_ci.helpers.dr_helpers_ui.clusters_in_dr_relationship(acm_obj, locator: tuple, timeout=30, expected_text=None)

This function is to verify there are 2 clusters in a healthy DR relationship

Parameters:

acm_obj (AcmAddClusters) – ACM Page Navigator Class
locator (tuple) – Locator for the element to be searched
timeout (int) – Timeout for which status check should be done
expected_text (str) – Text to be searched

Returns:

True if expected_text is found, False otherwise

Return type:

bool

ocs_ci.helpers.dr_helpers_ui.delete_application_ui(acm_obj, workloads_to_delete=[], timeout=70)

Function to delete specified workloads on ACM UI

Parameters:

acm_obj (AcmAddClusters) – ACM Page Navigator Class
workloads_to_delete (list) – Specify the workloads to delete
timeout (int) – timeout to wait for certain elements to be found on the ACM UI

Returns:

True if the application is deleted successfully, false otherwise

Return type:

bool

ocs_ci.helpers.dr_helpers_ui.dr_submariner_validation_from_ui(acm_obj)

This function is only applicable for Regional DR.

This function calls other function and does pre-checks on ACM UI such as Submariner validation from ACM console for Regional DR.

Parameters:: acm_obj (AcmAddClusters) – ACM Page Navigator Class

ocs_ci.helpers.dr_helpers_ui.failover_relocate_ui(acm_obj, scheduling_interval=0, workload_to_move=None, policy_name=None, failover_or_preferred_cluster=None, action='Failover', timeout=120, move_workloads_to_same_cluster=False, workload_type='Subscription', do_not_trigger=False)

Function to perform Failover/Relocate operations via ACM UI

Parameters:

acm_obj (AcmAddClusters) – ACM Page Navigator Class
scheduling_interval (int) – scheduling interval value from DRPolicy
workload_to_move (str) – Name of running workloads on which action to be taken
policy_name (str) – Name of the DR policy applied to the running workload
failover_or_preferred_cluster (str) – Name of the failover cluster or preferred cluster to which workloads will be moved
action (str) – action could be “Failover” or “Relocate”, “Failover” is set to default
timeout (int) – timeout to wait for certain elements to be found on the ACM UI
move_workloads_to_same_cluster (bool) – Bool condition to test negative failover/relocate scenarios to move running workloads to same cluster
workload_type (str) – Type of workload, appset or subscription
do_not_trigger (bool) – If in case you do not want to click on the Initiate button so as not to initiate the operation, set it to True. It’s False by default.

Returns:

True if the action is triggered, raises Exception if any of the mandatory argument is not provided

Return type:

bool

ocs_ci.helpers.dr_helpers_ui.health_and_peer_connection_check_on_ui(acm_obj, cluster1, cluster2, timeout=15, expected_text='Degraded')

The function checks the cluster and operator health, peer connection of both the managed clusters in a DR relationship

Parameters:

acm_obj (AcmAddClusters) – ACM Page Navigator Class
cluster1 (str) – Name of managed cluster one (primary preferably)
cluster2 (str) – Name of managed cluster two (secondary is most cases)
timeout (int) – Timeout for which the expected text would be checked
expected_text (str) – Text available on DR monitoring dashboard for Cluster and Operator status

Returns:

False if text Degraded is found either for cluster or operator health for any of the managed clusters, True if it is not found for both of them

ocs_ci.helpers.dr_helpers_ui.navigate_using_fleet_virtualization(acm_obj)

Starting ACM 2.15, VMs page from the ACM console has been removed and is integrated with the Virtulization Operator which is required to be installed on the ACM hub cluster and has it’s own perspective dropdown to switch to, which is called Fleet Virtulization.

This function is to navigate to the new VMs page using the Fleet Virtulization dropdown and connect dots with the existing tests so as to apply DR Policy to the CNV VM workloads from this page using Standalone or Shared Protection type.

Refer ACM-23371 and ACM-22068 for more details

Parameters:: acm_obj (AcmAddClusters) – ACM Page Navigator Class
Returns:: True if VM is found on the selected managed cluster and function executes successfully, False otherwise

ocs_ci.helpers.dr_helpers_ui.protected_volume_count_per_cluster(acm_obj, cluster_name)

Function to check total protected volume count on selected cluster :param cluster_name: Name of the managed cluster where apps are primary :type cluster_name: str

Returns:: DR protected total volume count on the selected cluster

ocs_ci.helpers.dr_helpers_ui.verify_application_present_in_ui(acm_obj, workloads_to_check=[], timeout=60)

Verify if application is present in UI

Parameters:

acm_obj (AcmAddClusters) – ACM Page Navigator Class
workloads_to_check (list) – Specify the workloads to check if they exist
timeout (int) – timeout to wait for certain elements to be found on the ACM UI

Returns:

True if the application is present, false otherwise

Return type:

bool

ocs_ci.helpers.dr_helpers_ui.verify_drpolicy_ui(acm_obj, scheduling_interval)

Function to verify DRPolicy status and replication policy on Data Policies page of ACM console

Parameters:

acm_obj (AcmAddClusters) – ACM Page Navigator Class
scheduling_interval (int) – Scheduling interval in the DRPolicy to be verified on ACM UI

ocs_ci.helpers.dr_helpers_ui.verify_failover_relocate_status_ui(acm_obj, action='Failover', timeout=120)

Function to verify current status of in progress Failover/Relocate operation on ACM UI

Parameters:

acm_obj (AcmAddClusters) – ACM Page Navigator Class
action (str) – action “Failover” or “Relocate” which was taken on the workloads, “Failover” is set to default
timeout (int) – timeout to wait for certain elements to be found on the ACM UI

ocs_ci.helpers.e2e_helpers module

ocs_ci.helpers.e2e_helpers.create_muliple_types_provider_obcs(num_of_buckets, bucket_types, cloud_providers, bucket_factory)

This function creates valid OBCs of different cloud providers and bucket types

Parameters:

num_of_buckets (int) – Number of buckets
bucket_types (dict) – Dict representing mapping between bucket type and relevant configuration
cloud_providers (dict) – Dict representing mapping between cloud providers and relevant configuration
bucket_factory (fixture) – bucket_factory fixture method

Returns:

list of created buckets

Return type:

List

ocs_ci.helpers.e2e_helpers.run_metadata_io_with_cephfs(dc_pod_factory, no_of_io_pods=3)

This function facilitates 1. Create PVC with Cephfs, access mode RWX 2. Create dc pod with Fedora image 3. Copy helper_scripts/meta_data_io.py to Fedora dc pod 4. Run meta_data_io.py on fedora pod

Parameters:

no_of_io_pods – by default 3 IO pods will be created. We can modify this by passing required value as argument.
dc_pod_factory – a fixture which will take care of dc pod creation and termination.

ocs_ci.helpers.e2e_helpers.validate_mcg_bucket_replicaton(awscli_pod_session, mcg_obj_session, source_target_map, uploaded_objects_dir, downloaded_obejcts_dir, event, run_in_bg=False, object_amount=5)

Validate MCG bucket replication feature

Parameters:

awscli_pod_session (Pod) – Pod object representing aws-cli pod
mcg_obj_session (MCG) – MCG object
source_target_map (Dict) – Dictionary consisting of source - target buckets
uploaded_objects_dir (str) – directory where uploaded objects are kept
downloaded_obejcts_dir (str) – directory where downloaded objects are kept
event (threading.Event()) – Event() object
run_in_bg (bool) – If True, validation is run in background
object_amount (int) – Amounts of objects

ocs_ci.helpers.e2e_helpers.validate_mcg_caching(awscli_pod_session, mcg_obj_session, cld_mgr, cache_buckets, uploaded_objects_dir, downloaded_obejcts_dir, event, run_in_bg=False)

Validate noobaa caching feature against the cache buckets

Parameters:

awscli_pod_session (Pod) – Pod object representing aws-cli pod
mcg_obj_session (MCG) – MCG object
cld_mgr (cld_mgr) – cld_mgr object
cache_buckets (List) – List consisting of cache buckets
uploaded_objects_dir (str) – directory where uploaded objects are kept
downloaded_obejcts_dir (str) – directory where downloaded objects are kept
event (threading.Event()) – Event() object
run_in_bg (bool) – If True, validation is run in background

ocs_ci.helpers.e2e_helpers.validate_mcg_nsfs_feature()

ocs_ci.helpers.e2e_helpers.validate_mcg_object_expiration(mcg_obj, buckets, event, run_in_bg=False, object_amount=5)

Validates objects expiration for MCG buckets

Parameters:

mcg_obj (MCG) – MCG object
buckets (List) – List of MCG buckets
event (threading.Event()) – Event() object
run_in_bg (Bool) – True if wants to run in background
object_amount (Int) – Amount of objects
prefix (str) – Any prefix used for objects

ocs_ci.helpers.e2e_helpers.validate_rgw_kafka_notification(kafka_rgw_dict, event, run_in_bg=False)

Validate kafka notifications for RGW buckets

Parameters:

kafka_rgw_dict (Dict) – Dict consisting of rgw bucket,
kafka_topic –
etc (kafkadrop_host) –
event (threading.Event()) – Event() object
run_in_bg (Bool) – True if you want to run in the background

ocs_ci.helpers.e2e_helpers.verify_osd_used_capacity_greater_than_expected(expected_used_capacity)

Verify OSD percent used capacity greate than ceph_full_ratio

Parameters:: expected_used_capacity (float) – expected used capacity
Returns:: True if used_capacity greater than expected_used_capacity, False otherwise
Return type:: bool

ocs_ci.helpers.github module

ocs_ci.helpers.github.get_asset_from_github(name, owner_repo, release_tag='latest')

Download and install asset from github.

Parameters:

name (str) – name of the tool which should be downloaded
owner_repo (str) – github repository with the tool in form owner/repo
release_tag (str) – release tag to download (default: latest)

ocs_ci.helpers.helpers module

Helper functions file for OCS QE

ocs_ci.helpers.helpers.add_route_public_nad()

Add route section to network_attachment_definitions object

Adds route to shim network for host-to-pod communication in VLAN mode. The shim network is configured via ‘multus_public_net_shim_network’ ENV_DATA parameter (default: 192.168.252.0/24). This route enables communication between baremetal hosts and pods attached to the public Multus network.

ocs_ci.helpers.helpers.add_scc_policy(sa_name, namespace)

Adding ServiceAccount to scc anyuid and privileged

Parameters:

sa_name (str) – ServiceAccount name
namespace (str) – The namespace for the scc_policy creation

ocs_ci.helpers.helpers.apply_custom_taint_and_toleration(taint_label='xyz')

Apply custom taints and tolerations. 1. Taint ocs nodes with non-ocs taint 2. Set custom tolerations on storagecluster, subscription, configmap, ocsinit and drivers

Parameters:: taint_label (str) – The taint label to apply (default is “xyz”).

ocs_ci.helpers.helpers.apply_resource(**kwargs)

Apply a resource. Safe for both create and update operations.

Parameters:: kwargs (dict) – Dictionary of the OCS resource
Returns:: An OCS instance
Return type:: OCS
Raises:: AssertionError – In case of any failure

ocs_ci.helpers.helpers.calc_local_file_md5_sum(path)

Calculate and return the MD5 checksum of a local file

Parameters:: path (str) – The path to the file
Returns:: The MD5 checksum
Return type:: str

ocs_ci.helpers.helpers.ceph_health_check_with_toolbox_recovery(namespace: str, tries: int = 20, delay: int = 30, fix_ceph_health: bool = True, update_jira: bool = True, no_exception_if_jira_issue_updated: bool = False) → bool

Perform ceph health check with automatic toolbox pod recovery.

If the ceph toolbox pod is not running (e.g., after node disruption tests), this function will wait for the toolbox pod to recover before retrying.

Parameters:

namespace (str) – Kubernetes namespace for ceph cluster.
tries (int) – Number of retries for health check.
delay (int) – Delay between retries in seconds.
fix_ceph_health (bool) – Whether to attempt fixing ceph health issues.
update_jira (bool) – Whether to update Jira on health issues.
no_exception_if_jira_issue_updated (bool) – Skip exception if Jira was updated.

Returns:

True if ceph health check passes.

Return type:

bool

Raises:

NoRunningCephToolBoxException – If toolbox pod doesn’t recover.
CephHealthException – If ceph health check fails after retries.

ocs_ci.helpers.helpers.change_default_storageclass(scname)

Change the default StorageClass to the given SC name

Parameters:: scname (str) – StorageClass name
Returns:: True on success
Return type:: bool

ocs_ci.helpers.helpers.change_reclaimspacecronjob_state_for_pvc(pvc_objs, suspend=True)

Enable or disable the ReclaimSpace operation for the PVC’s ReclaimSpaceCronJob.

Parameters:

pvc_objs (list) – List of PersistentVolumeClaim (PVC) objects.
suspend (bool) – If True, disables ReclaimSpace; if False, enables ReclaimSpace.

Returns:

True if the operation was successfully applied to all PVCs.

Return type:

bool

ocs_ci.helpers.helpers.change_vm_network_state(ip, label='Network adapter 1', network='VM Network', connect=False)

Changes the network state of a virtual machine.

Parameters:

ip (str) – The IP address of the virtual machine.
label (str, optional) – The label of the network adapter to be changed. Defaults to constants.VM_DEFAULT_NETWORK_ADAPTER.
network (str, optional) – The name of the network to which the network adapter should be connected. Defaults to constants.VM_DEFAULT_NETWORK.
connect (bool, optional) – If True, the network adapter is connected. If False, the network adapter is disconnected. Defaults to False.

Returns:

Returns True if the operation was successful, False otherwise.

Return type:

bool

ocs_ci.helpers.helpers.check_cluster_is_compact()

ocs_ci.helpers.helpers.check_number_of_mon_pods(expected_mon_num=3)

Function to check the number of monitoring pods

Returns:: True if number of mon pods is 3, False otherwise
Return type:: bool

ocs_ci.helpers.helpers.check_osd_log_exist_on_rook_ceph_operator_pod(last_log_date_time_obj, expected_strings=(), unexpected_strings=())

Verify logs contain the expected strings and the logs do not: contain the unexpected strings

Parameters:

last_log_date_time_obj (datetime obj) – type of log
expected_strings (list) – verify the logs contain the expected strings
unexpected_strings (list) – verify the logs do not contain the strings

Returns:

True if logs contain the expected strings and the logs do not contain the unexpected strings, False otherwise

Return type:

bool

ocs_ci.helpers.helpers.check_osds_down(osd_ids: list[str]) → bool

Check if specified OSDs are marked as ‘down’ in Ceph

Parameters:: osd_ids (list[str]) – List of OSD IDs to check
Returns:: True if all OSDs are down, False otherwise
Return type:: bool

ocs_ci.helpers.helpers.check_pods_status_by_pattern(pattern, namespace, expected_status)

Check if the pod state is as expected.

Parameters:

pattern (str) –
namespace (str) –
expected_status (str) –

Returns:

return True if pod in expected status otherwise False

Return type:

bool

ocs_ci.helpers.helpers.check_rbd_image_used_size(pvc_objs, usage_to_compare, rbd_pool=None, expect_match=True)

Check if RBD image used size of the PVCs are matching with the given value

Parameters:

pvc_objs (list) – List of PVC objects
usage_to_compare (str) – Value of image used size to be compared with actual value. eg: “5GiB”
rbd_pool (str) – Name of the RBD metadata pool (parameters.pool from the StorageClass). On EC clusters this is the replicated metadata pool (e.g. replicated-metadata-pool), not the EC data pool. When None the value is resolved automatically from the default RBD StorageClass via default_ceph_block_pool().
expect_match (bool) – True to verify the used size is equal to ‘usage_to_compare’ value. False to verify the used size is not equal to ‘usage_to_compare’ value.

Returns:

True if the verification is success for all the PVCs, False otherwise

Return type:

bool

ocs_ci.helpers.helpers.check_rook_ceph_crashcollector_pods_where_rook_ceph_pods_are_running()

check rook-ceph-crashcollector pods running on worker nodes where rook-ceph pods are running.

Returns:

True if the rook-ceph-crashcollector pods running on worker nodes: where rook-ceph pods are running. False otherwise.

Return type:

bool

ocs_ci.helpers.helpers.check_selinux_relabeling(pod_obj)

Check SeLinux Relabeling is set to false.

Parameters:: pod_obj (Pod object) – App pod

ocs_ci.helpers.helpers.clean_all_test_projects(project_name='test')

Delete all namespaces with ‘test’ in its name ‘test’ can be replaced with another string

Parameters:: project_name (str) – expression to be deleted. Defaults to “test”.

ocs_ci.helpers.helpers.clear_crash_warning_and_osd_removal_leftovers(): Clear crash warnings and osd removal leftovers. This function can be used for example, after the device replacement test or the node replacement test.

ocs_ci.helpers.helpers.collect_performance_stats(dir_name)

Collect performance stats and saves them in file in json format.

dir_name (str): directory name to store stats.

Performance stats include:: IOPs and throughput percentage of cluster CPU, memory consumption of each nodes

ocs_ci.helpers.helpers.configure_cephcluster_params_in_storagecluster_cr(params, default_values=False)

Configure cephcluster block in StorageCluster CR /spec/managedResources/cephCluster/

Parameters:

params (list) – A list of dictionaries with value for cephCluster in StorageCluster CR
default_values (bool) – parameters to set in StorageCluster under /spec/managedResources/cephCluster/

ocs_ci.helpers.helpers.configure_node_network_configuration_policy_on_all_worker_nodes()

Configure NodeNetworkConfigurationPolicy CR on each worker node in cluster

Supports both traditional shim-based approach and VLAN-based approach. Use multus_use_vlan config parameter to enable VLAN mode.

ocs_ci.helpers.helpers.converge_lists(list_to_converge)

Function to flatten and remove the sublist created during future obj

Parameters:: list_to_converge (list) – arg list of lists, eg: [[1,2],[3,4]]
Returns:: return converged list eg: [1,2,3,4]
Return type:: list (list)

ocs_ci.helpers.helpers.create_auto_scaler(name=None, namespace=None, sc_name=None, device_class=None, capacity_limit='8Ti', scaling_threshold=70, max_osd_size='8Ti', timeout=1800)

Create a StorageAutoScaler custom resource in OpenShift.

Parameters:

name (str) – Name of the StorageAutoScaler resource.
namespace (str) – Namespace where the resource is created.
sc_name (str) – Name of the StorageCluster to attach to.
device_class (str) – Device class for OSDs.
capacity_limit (str) – Maximum total capacity before scaling stops.
scaling_threshold (int) – Percent usage to trigger auto-scaling.
max_osd_size (str) – Size of each OSD added during scaling.
timeout (int) – Timeout in seconds for a scaling operation.

Returns:

An OCS instance of the StorageAutoScaler

Return type:

OCS

ocs_ci.helpers.helpers.create_build_from_docker_image(image_name, install_package, namespace, source_image='quay.io/ocsci/fedora', source_image_label='fio')

Allows to create a build config using a Dockerfile specified as an argument, eg.:

$ oc new-build -D $'FROM centos:7\nRUN yum install -y httpd'

creates a build with httpd installed.

Parameters:

image_name (str) – Name of the image to be created
source_image (str) – Source image to build docker image from, defaults to Centos as base image
namespace (str) – project where build config should be created
source_image_label (str) – Tag to use along with the image name, defaults to ‘latest’
install_package (str) – package to install over the base image

Returns:

The OCP object for the image Fails on UnavailableBuildException exception if build creation fails

Return type:

ocs_ci.ocs.ocp.OCP (obj)

ocs_ci.helpers.helpers.create_ceph_block_pool(pool_name=None, replica=3, compression=None, failure_domain=None, verify=True, namespace=None, device_class=None, yaml_file=None)

Create a Ceph block pool with optional parameters.

Parameters:

pool_name (str) – The pool name to create (optional).
replica (int) – The replica size for the pool.
compression (str) – Compression type for the pool (optional).
failure_domain (str) – Failure domain name (optional).
verify (bool) – True to verify the pool exists after creation, False otherwise.
namespace (str) – The pool namespace (optional).
device_class (str) – The device class name (optional).
yaml_file (str) – The name of the YAML file for the Ceph block pool (optional).

Returns:

The OCS instance for the Ceph block pool.

Return type:

OCS

ocs_ci.helpers.helpers.create_ceph_file_system(cephfs_name=None, label=None, namespace=None)

Create a Ceph file system

Parameters:

cephfs_name (str) – The ceph FS name to create
label (dict) – The label to give to pool
namespace (str) – The name space in which the ceph FS has to be created

Returns:

An OCS instance for the Ceph file system

Return type:

OCS

ocs_ci.helpers.helpers.create_lvs_resource(name, storageclass, worker_nodes=None, min_size=None, max_size=None)

Create the LocalVolumeSet resource.

Parameters:

name (str) – The name of the LocalVolumeSet CR
storageclass (str) – storageClassName value to be used in LocalVolumeSet CR based on LOCAL_VOLUME_YAML
worker_nodes (list) – The worker node names to be used in the LocalVolumeSet resource
min_size (str) – The min size to be used in the LocalVolumeSet resource
max_size (str) – The max size to be used in the LocalVolumeSet resource

Returns:

The OCS instance for the LocalVolumeSet resource

Return type:

OCS

ocs_ci.helpers.helpers.create_multilpe_projects(number_of_project)

Create one or more projects

Parameters:: number_of_project (int) – Number of projects to be created
Returns:: List of project objects
Return type:: list

ocs_ci.helpers.helpers.create_multiple_pvc_parallel(sc_obj, namespace, number_of_pvc, size, access_modes)

Funtion to create multiple PVC in parallel using threads Function will create PVCs based on the available access modes

Parameters:

sc_obj (str) – Storage Class object
namespace (str) – The namespace for creating pvc
number_of_pvc (int) – NUmber of pvc to be created
size (str) – size of the pvc eg: ‘10Gi’
access_modes (list) – List of access modes for PVC creation

Returns:

List of pvc objs created in function

Return type:

pvc_objs_list (list)

ocs_ci.helpers.helpers.create_multiple_pvcs(sc_name, namespace, number_of_pvc=1, size=None, do_reload=False, access_mode='ReadWriteOnce', burst=False)

Create one or more PVC as a bulk or one by one

Parameters:

sc_name (str) – The name of the storage class to provision the PVCs from
namespace (str) – The namespace for the PVCs creation
number_of_pvc (int) – Number of PVCs to be created
size (str) – The size of the PVCs to create
do_reload (bool) – True for wait for reloading PVC after its creation, False otherwise
access_mode (str) – The kind of access mode for PVC
burst (bool) – True for bulk creation, False ( default) for multiple creation

Returns:

List of PVC objects tmpdir (str): The full path of the directory in which the yamls for pvc objects creation reside

Return type:

ocs_objs (list)

ocs_ci.helpers.helpers.create_network_fence(node_name, cidr)

Create NetworkFence for the node

Parameters:

node_name (str) – Name of the node
cidr (str) – cidr

Returns:

NetworkFence object

Return type:

OCS

ocs_ci.helpers.helpers.create_network_fence_class(): Create NetworkFenceClass CR and verify Ips are populated in respective CsiAddonsNode objects

ocs_ci.helpers.helpers.create_ocs_object_from_kind_and_name(kind, resource_name, namespace=None)

Create OCS object from kind and name

Parameters:

kind (str) – resource kind like CephBlockPool, pvc.
resource_name (str) – name of the resource.
namespace (str) – then value from config will be used.

Returns:

returns OCS object from kind and name.

Return type:

ocs_ci.ocs.resources.ocs.OCS (obj)

ocs_ci.helpers.helpers.create_pod(interface_type=None, pvc_name=None, do_reload=True, namespace=None, node_name=None, pod_dict_path=None, sa_name=None, security_context=None, raw_block_pv=False, raw_block_device='/dev/rbdblock', replica_count=1, pod_name=None, node_selector=None, command=None, command_args=None, ports=None, subpath=None, deployment=False, scc=None, volumemounts=None, pvc_read_only_mode=None, priorityClassName=None)

Create a pod

Parameters:

interface_type (str) – The interface type (CephFS, RBD, etc.)
pvc_name (str) – The PVC that should be attached to the newly created pod
do_reload (bool) – True for reloading the object after creation, False otherwise
namespace (str) – The namespace for the new resource creation
node_name (str) – The name of specific node to schedule the pod
pod_dict_path (str) – YAML path for the pod
sa_name (str) – Serviceaccount name
security_context (dict) – Set security context on container in the form of dictionary
raw_block_pv (bool) – True for creating raw block pv based pod, False otherwise
raw_block_device (str) – raw block device for the pod
replica_count (int) – Replica count for deployment config
pod_name (str) – Name of the pod to create
node_selector (dict) – dict of key-value pair to be used for nodeSelector field eg: {‘nodetype’: ‘app-pod’}
command (list) – The command to be executed on the pod
command_args (list) – The arguments to be sent to the command running on the pod
ports (dict) – Service ports
subpath (str) – Value of subPath parameter in pod yaml
deployment (bool) – True for Deployment creation, False otherwise
scc (dict) – Set security context on pod like fsGroup, runAsUer, runAsGroup
volumemounts (list) – Value of mountPath parameter in pod yaml

Returns:

A Pod instance

Return type:

Pod

Raises:

AssertionError – In case of any failure

ocs_ci.helpers.helpers.create_pods(pvc_objs, pod_factory, interface, pods_for_rwx=1, status='', nodes=None)

Create pods

Parameters:

pvc_objs (list) – List of ocs_ci.ocs.resources.pvc.PVC instances
pod_factory (function) – pod_factory function
interface (int) – Interface type
pods_for_rwx (int) – Number of pods to be created if access mode of PVC is RWX
status (str) – If provided, wait for desired state of each pod before creating next one
nodes (list) – Node name for each pod will be selected from this list.

Returns:

list of Pod objects

Return type:

list

ocs_ci.helpers.helpers.create_pods_parallel(pvc_list, namespace, interface, pod_dict_path=None, sa_name=None, raw_block_pv=False, deployment=False, node_selector=None)

Function to create pods in parallel

Parameters:

pvc_list (list) – List of pvcs to be attached in pods
namespace (str) – The namespace for creating pod
interface (str) – The interface backed the PVC
pod_dict_path (str) – pod_dict_path for yaml
sa_name (str) – sa_name for providing permission
raw_block_pv (bool) – Either RAW block or not
deployment (bool) – True for Deployment creation, False otherwise
node_selector (dict) – dict of key-value pair to be used for nodeSelector field eg: {‘nodetype’: ‘app-pod’}

Returns:

Returns list of pods created

Return type:

pod_objs (list)

ocs_ci.helpers.helpers.create_priority_class(priority, value): Function to create priority class on the cluster :returns: Returns priority class obj :rtype: bool

ocs_ci.helpers.helpers.create_project(project_name=None)

Create a project

Parameters:: project_name (str) – The name for the new project
Returns:: Project object
Return type:: ocs_ci.ocs.ocp.OCP

ocs_ci.helpers.helpers.create_pvc(sc_name, pvc_name=None, namespace=None, size=None, do_reload=True, access_mode='ReadWriteOnce', volume_mode=None, volume_name=None)

Create a PVC

Parameters:

sc_name (str) – The name of the storage class for the PVC to be associated with
pvc_name (str) – The name of the PVC to create
namespace (str) – The namespace for the PVC creation
size (str) – Size of pvc to create
do_reload (bool) – True for wait for reloading PVC after its creation, False otherwise
access_mode (str) – The access mode to be used for the PVC
volume_mode (str) – Volume mode for rbd RWX pvc i.e. ‘Block’
volume_name (str) – Persistent Volume name

Returns:

PVC instance

Return type:

PVC

ocs_ci.helpers.helpers.create_rbd_deviceclass_storageclass(pool_name, sc_name=None, cluster_id='openshift-storage', reclaim_policy='Delete', volume_binding_mode='WaitForFirstConsumer', image_features=None, encrypted='false', allow_volume_expansion=True)

Create an RBD StorageClass resource for device class from provided parameters.

Parameters:

pool_name (str) – Name of the pool.
sc_name (str) – Name of the StorageClass. If not provided, it will set a random name.
cluster_id (str) – Cluster ID.
reclaim_policy (str) – Reclaim policy (e.g., “Delete” or “Retain”).
volume_binding_mode (str) – Volume binding mode (e.g., “Immediate”, “WaitForFirstConsumer”).
image_features (str) – Image features for the pool.
encrypted (str) – Encryption flag (“true” or “false”).
allow_volume_expansion (bool) – Allow volume expansion (True/False).

Returns:

The OCS instance for the StorageClass resource

Return type:

OCS

ocs_ci.helpers.helpers.create_reclaim_space_cronjob(pvc_name, reclaim_space_job_name=None, backoff_limit=None, retry_deadline_seconds=None, schedule='weekly')

Create ReclaimSpaceCronJob to invoke reclaim space operation on RBD volume

Parameters:

pvc_name (str) – Name of the PVC
reclaim_space_job_name (str) – The name of the ReclaimSpaceCRonJob to be created
backoff_limit (int) – The number of retries before marking reclaim space operation as failed
retry_deadline_seconds (int) – The duration in seconds relative to the start time that the operation may be retried
schedule (str) – Type of schedule

Returns:

An OCS object representing ReclaimSpaceJob

Return type:

ocs_ci.ocs.resources.ocs.OCS

ocs_ci.helpers.helpers.create_reclaim_space_job(pvc_name, reclaim_space_job_name=None, backoff_limit=None, retry_deadline_seconds=None)

Create ReclaimSpaceJob to invoke reclaim space operation on RBD volume

Parameters:

pvc_name (str) – Name of the PVC
reclaim_space_job_name (str) – The name of the ReclaimSpaceJob to be created
backoff_limit (int) – The number of retries before marking reclaim space operation as failed
retry_deadline_seconds (int) – The duration in seconds relative to the start time that the operation may be retried

Returns:

An OCS object representing ReclaimSpaceJob

Return type:

ocs_ci.ocs.resources.ocs.OCS

ocs_ci.helpers.helpers.create_resource(do_reload=True, **kwargs)

Create a resource

Parameters:

do_reload (bool) – True for reloading the resource following its creation, False otherwise
kwargs (dict) – Dictionary of the OCS resource

Returns:

An OCS instance

Return type:

OCS

Raises:

AssertionError – In case of any failure

ocs_ci.helpers.helpers.create_sa_token_secret(sa_name, namespace=None)

Creates a serviceaccount token secret

Parameters:

sa_name (str) – Name of the serviceaccount for which the secret has to be created
namespace (str) – Namespace in which the serviceaccount exists

Returns:

Name of the serviceaccount token secret

Return type:

str

ocs_ci.helpers.helpers.create_scc(scc_name=None, scc_dict=None, scc_dict_path=None)

Create a SecurityContextConstraints

Parameters:

scc_name (str) – Name of the SCC
scc_dict (dict) – Dictionary containing the details on provileges, capabilities etc
scc_dict_path (str) – Path to custom SCC yaml file

Returns:

OCS object for scc created

Return type:

scc_obj

ocs_ci.helpers.helpers.create_secret(interface_type)

Create a secret ** This method should not be used anymore ** ** This method is for internal testing only **

Parameters:: interface_type (str) – The type of the interface (e.g. CephBlockPool, CephFileSystem)
Returns:: An OCS instance for the secret
Return type:: OCS

ocs_ci.helpers.helpers.create_serviceaccount(namespace)

Create a Serviceaccount

Parameters:: namespace (str) – The namespace for the serviceaccount creation
Returns:: An OCS instance for the service_account
Return type:: OCS

ocs_ci.helpers.helpers.create_storage_class(interface_type, interface_name, secret_name, reclaim_policy='Delete', sc_name=None, provisioner=None, rbd_thick_provision=False, encrypted=False, encryption_kms_id=None, fs_name=None, volume_binding_mode='Immediate', allow_volume_expansion=True, kernelMountOptions=None, annotations=None, mapOptions=None, mounter=None)

Create a storage class ** This method should not be used anymore ** ** This method is for internal testing only **

Parameters:

interface_type (str) – The type of the interface (e.g. CephBlockPool, CephFileSystem)
interface_name (str) – The name of the interface
secret_name (str) – The name of the secret
sc_name (str) – The name of storage class to create
reclaim_policy (str) – Type of reclaim policy. Defaults to ‘Delete’ (eg., ‘Delete’, ‘Retain’)
rbd_thick_provision (bool) – True to enable RBD thick provisioning. Applicable if interface_type is CephBlockPool
encrypted (bool) – True to create encrypted SC else False
encryption_kms_id (str) – ID of the KMS entry from connection details
fs_name (str) – the name of the filesystem for CephFS StorageClass
volume_binding_mode (str) – Can be “Immediate” or “WaitForFirstConsumer” which the PVC will be in pending till pod attachment.
allow_volume_expansion (bool) – True to create sc with volume expansion
kernelMountOptions (str) – Mount option for security context
annotations (dict) – dict of annotations to be added to the storageclass.
mapOptions (str) – mapOtions match the configuration of ocs-storagecluster-ceph-rbd-virtualization storage class
mounter (str) – mounter to match the configuration of ocs-storagecluster-ceph-rbd-virtualization storage class

Returns:

An OCS instance for the storage class

Return type:

OCS

ocs_ci.helpers.helpers.create_unique_resource_name(resource_description, resource_type)

Creates a unique object name by using the object_description, object_type and a random uuid(in hex) as suffix trimmed due to kubernetes limitation of 63 characters

Parameters:

resource_description (str) – The user provided object description
resource_type (str) – The type of object for which the unique name will be created. For example: project, pvc, etc

Returns:

A unique name

Return type:

str

ocs_ci.helpers.helpers.default_ceph_block_pool()

Returns default CephBlockPool

Returns:: default CephBlockPool

ocs_ci.helpers.helpers.default_storage_class(interface_type)

Return default storage class based on interface_type

This function usually runs as one of the first functions after the StorageCluster CR is applied. This makes the function more prone to failures when DF/DF Client operators have not finalized provisioning; retry adds redundancy to the function.

Parameters:: interface_type (str) – The type of the interface (e.g. CephBlockPool, CephFileSystem)
Returns:: Existing StorageClass Instance
Return type:: OCS

ocs_ci.helpers.helpers.default_thick_storage_class()

Return default RBD thick storage class

Returns:: Existing RBD thick StorageClass instance
Return type:: OCS

ocs_ci.helpers.helpers.default_volumesnapshotclass(interface_type)

Return default VolumeSnapshotClass based on interface_type

Parameters:: interface_type (str) – The type of the interface (e.g. CephBlockPool, CephFileSystem)
Returns:: VolumeSnapshotClass Instance
Return type:: OCS

ocs_ci.helpers.helpers.delete_bulk_pvcs(pvc_yaml_dir, pv_names_list, namespace): Deletes all the pvcs created from yaml file in a provided dir :param pvc_yaml_dir: Directory in which yaml file resides :type pvc_yaml_dir: str :param pv_names_list: List of pv objects to be deleted :type pv_names_list: str

ocs_ci.helpers.helpers.delete_cephblockpools(cbp_objs)

Function for deleting CephBlockPool

Parameters:: cbp_objs (list) – List of CBP objects for deletion
Returns:: True if deletion of CephBlockPool is successful
Return type:: bool

ocs_ci.helpers.helpers.delete_csi_holder_daemonsets(): Delete csi holder daemonsets

ocs_ci.helpers.helpers.delete_csi_holder_pods()

Drain/schedule worker nodes and reset csi-holder-pods

Procedure: 1.Cordon worker node-X 2.Drain worker node-X 3.Reset csi-cephfsplugin-holder and csi-rbdplugin-holder pods on node-X 4.schedule node-X 5.Verify all node-X in Ready state

ocs_ci.helpers.helpers.delete_objs_parallel(obj_list)

Function to delete objs specified in list :param obj_list: List can be obj of pod, pvc, etc :type obj_list: list

Returns:: True if obj deleted else False
Return type:: bool

ocs_ci.helpers.helpers.delete_storageclasses(sc_objs)

Function for Deleting storageclasses

Parameters:: sc_objs (list) – List of SC objects for deletion
Returns:: True if deletion is successful
Return type:: bool

ocs_ci.helpers.helpers.delete_volume_in_backend(img_uuid, pool_name=None, disable_mirroring=False)

Delete an Image/Subvolume in the backend

Parameters:

img_uuid (str) – Part of VolID which represents corresponding image/subvolume in backend, eg: oc get pv/<volumeName> -o jsonpath='{.spec.csi.volumeHandle}' Output is the CSI generated VolID and looks like: 0001-000c-rook-cluster-0000000000000001-f301898c-a192-11e9-852a-1eeeb6975c91 where image_uuid is f301898c-a192-11e9-852a-1eeeb6975c91
pool_name (str) – The name of the pool
disable_mirroring (bool) – True to disable the mirroring for the image, False otherwise

Returns:

True if image deleted successfully

False if:: Pool not found image not found image not deleted

Return type:

bool

ocs_ci.helpers.helpers.disable_vm_network_for_duration(ip, label='Network adapter 1', network='VM Network', duration=5)

Disable network connectivity for a virtual machine with a specified IP address for a given duration.

Parameters:

ip (str) – The IP address of the virtual machine to disable network connectivity for.
label (str, optional) – The label of the network adapter to disable. (default: “Network adapter 1”)
network (str, optional) – The name of the network to connect to. (default: “VM Network”)
duration (int, optional) – The duration in seconds to disable network connectivity. (default: 5 seconds)

Returns:

True if network connectivity was successfully disabled and re-enabled, False otherwise.

Return type:

bool

ocs_ci.helpers.helpers.enable_csi_disable_holder_pods(): Enable CSI_DISABLE_HOLDER_PODS in rook-ceph-operator-config config-map

ocs_ci.helpers.helpers.fetch_used_size(cbp_name, exp_val=None)

Fetch used size in the pool :param exp_val: Expected size in GB :type exp_val: float

Returns:: Used size in GB
Return type:: float

ocs_ci.helpers.helpers.find_cephblockpoolradosnamespace(storageclient_uid=None)

Find the cephblockpoolradosnamespace related to a storageclient. This should run on provider cluster in a: provider mode setup.

! Important. from 4.19 onwards, the StorageRequest is deprecated. ! rns will be fetched from configmap of storageConsumer

Parameters:: storageclient_id (string) – The uid of the storageclient for which the cephblockpoolradosnamespace to be fetched
Returns:: The name of the cephblockpoolradosnamespace, if present
Return type:: str

ocs_ci.helpers.helpers.find_cephfilesystemsubvolumegroup(storageclient_uid=None)

Find the cephfilesystemsubvolumegroup related to a storageclient. This should run on provider cluster in a: provider mode setup.

! Important. from 4.19 onwards, the StorageRequest is deprecated. ! cephfilesystemsubvolumegroup will be fetched from configmap of storageConsumer

Parameters:: storageclient_id (string) – The uid of the storageclient for which the cephfilesystemsubvolumegroup to be fetched
Returns:: The name of the cephfilesystemsubvolumegroup, if present
Return type:: str

ocs_ci.helpers.helpers.find_radosnamespace(storageclient_uid=None)

Find the radosnamespace related to a storageclient if present

Parameters:: storageclient_id (string) – The uid of the storageclient for which the lradosnamespace has to be identified
Returns:: The name of the radosnamespace, if present
Return type:: str

ocs_ci.helpers.helpers.flatten_multilevel_dict(d)

Recursively extracts the leaves of a multi-level dictionary and returns them as a list.

Parameters:: d (dict) – The multi-level dictionary.
Returns:: A list containing the leaves of the dictionary.
Return type:: list

ocs_ci.helpers.helpers.get_admin_key()

Fetches admin key secret from Ceph

Returns:: The admin key
Return type:: str

ocs_ci.helpers.helpers.get_all_pvs()

Gets all pvs in cluster namespace (openshift-storage or fusion-storage)

Returns:: Dict of all pv in the cluster namespace
Return type:: dict

ocs_ci.helpers.helpers.get_all_storageclass_names()

Function for getting all storageclass

Returns:: list of storageclass name
Return type:: list

ocs_ci.helpers.helpers.get_architecture_path(cli_type)

Get Architcture path

Parameters:: cli_type (str) – choose which bin file you want to download [“odf” -> odf-cli , “mcg” -> mcg-cli]
Returns:: path of MCG/ODF CLI Binary in the image.
Return type:: (str)

ocs_ci.helpers.helpers.get_ceph_log_level(service, subsystem)

Return CEPH log level value.

Parameters:

service (_type_) – _description_
subsystem (_type_) – _description_

ocs_ci.helpers.helpers.get_cephblockpool_names()

Function for getting all CephBlockPool

Returns:: list of cephblockpool name
Return type:: list

ocs_ci.helpers.helpers.get_cephfs_data_pool_name()

Fetches ceph fs datapool name from Ceph

Returns:: fs datapool name
Return type:: str

ocs_ci.helpers.helpers.get_cephfs_name(): Function to retrive CephFS name :returns: Name of CFS :rtype: str

ocs_ci.helpers.helpers.get_cephfs_sc_name()

Get the cephfs storage class name.

Returns:: The cephfs storage class name.
Return type:: str
Raises:: ValueError – If the cephfs storage class name hasn’t been found.

ocs_ci.helpers.helpers.get_cephfs_subvolumegroup()

Get the name of cephfilesystemsubvolumegroup. The name should be fetched if the platform is not MS.

Returns:: The name of cephfilesystemsubvolumegroup
Return type:: str

ocs_ci.helpers.helpers.get_cephfs_subvolumegroup_names()

Get all CephFS subvolume groups present in the cluster.

Returns:: Names of all CephFS subvolume groups
Return type:: list

ocs_ci.helpers.helpers.get_current_test_name(): A function to return the current test name in a parsed manner :returns: The test name. :rtype: str

ocs_ci.helpers.helpers.get_daemonsets_names(namespace=None)

Get all daemonspaces in namespace

Parameters:: namespace (str) – namespace
Returns:: all daemonset names in the namespace
Return type:: list

ocs_ci.helpers.helpers.get_daemonsets_obj(name, namespace=None)

Get daemonset obj :param name: the name of daemeonset :type name: str :param namespace: the namespace of daemonset :type namespace: str

Returns:: daemonset ocp obj
Return type:: ocp_obj

ocs_ci.helpers.helpers.get_data_pool_name(interface_type='CephBlockPool', sc_obj=None)

Return the name of the pool where actual block data is written.

On standard (replicated) deployments the StorageClass has a single pool parameter that serves both as the RBD metadata pool and the data pool, so this function returns the same value as default_ceph_block_pool().

On Erasure-Coded deployments the StorageClass exposes two parameters:

pool – the replicated metadata pool (stores RBD image headers)
dataPool – the EC pool where all actual object data is written

For size-tracking purposes (fetch_used_size) we must monitor dataPool on EC clusters; monitoring the metadata pool would always show negligible growth regardless of how much data is written.

Parameters:

interface_type (str) – Interface type constant (default CEPHBLOCKPOOL).
sc_obj – Optional StorageClass OCP object. When provided, it is used directly instead of resolving the default StorageClass for interface_type. Useful for callers that already have a specific SC object (e.g. a custom pool created by a test).

Returns:

Name of the pool that receives the actual written data.

Return type:

str

ocs_ci.helpers.helpers.get_default_cluster_volumesnapshotclass()

Get the default VolumeSnapshotClass available in the OCS cluster.

This helper function retrieves the VolumeSnapshotClass that is marked as default using the annotation ‘snapshot.storage.kubernetes.io/is-default-class: “true”’. If a driver is specified, it will return the default snapshot class for that driver.

Returns:: The name of the default VolumeSnapshotClass, or None if not found
Return type:: str
Raises:: ResourceNotFoundError – If no default VolumeSnapshotClass is found

Examples

>>> # Get the default volume snapshot class
>>> default_snapclass = get_default_volumesnapshotclass()
>>> print(default_snapclass)
'vpc-block-snapshot'

ocs_ci.helpers.helpers.get_default_storage_class()

Get the default StorageClass(es)

Returns:: default StorageClass(es) list
Return type:: list

ocs_ci.helpers.helpers.get_end_creation_time(interface, pvc_name)

Get the ending creation time of a PVC based on provisioner logs

Parameters:

interface (str) – The interface backed the PVC
pvc_name (str) – Name of the PVC for creation time measurement

Returns:

End time of PVC creation

Return type:

datetime object

ocs_ci.helpers.helpers.get_end_deletion_time(interface, pv_name)

Get the ending deletion time of a PVC based on provisioner logs

Parameters:

interface (str) – The interface backed the PVC
pv_name (str) – Name of the PVC for deletion time measurement

Returns:

End time of PVC deletion

Return type:

datetime object

ocs_ci.helpers.helpers.get_event_line_datetime(event_line)

Get the event line datetime

Parameters:: event_line (str) – The event line to get it’s datetime
Returns:: The event line datetime
Return type:: datetime object

ocs_ci.helpers.helpers.get_failure_domain()

Get Failure Domain

Returns:: type of failure domain
Return type:: string

ocs_ci.helpers.helpers.get_failure_domin()

Function is used to getting failure domain of pool

Returns:: Failure domain from cephblockpool
Return type:: str

ocs_ci.helpers.helpers.get_full_test_logs_path(cname, fname=None)

Getting the full path of the logs file for particular test

this function use the inspect module to find the name of the caller function, so it need to be call once from the main test function. the output is in the form of ocsci_log_path/<full test file path>/<test filename>/<test class name>/<test function name>

Parameters:

cname (obj) – the Class object which was run and called this function
fname (str) – the function name for different tests log path

Returns:

full path of the test logs relative to the ocs-ci base logs path

Return type:

str

ocs_ci.helpers.helpers.get_last_log_time_date()

Get last log time

Returns:: type of log
Return type:: last_log_date_time_obj (datetime obj)

ocs_ci.helpers.helpers.get_logs_rook_ceph_operator()

Get logs from a rook_ceph_operator pod

Returns:: Output from ‘oc get logs rook-ceph-operator command
Return type:: str

ocs_ci.helpers.helpers.get_logs_with_errors(errors=None)

From logs of all pods and nodes, get only logs containing any of specified errors

Parameters:: errors (list) – List of errors to look for
Returns:: node/pod name as key, logs content as value; may be empty
Return type:: dict

ocs_ci.helpers.helpers.get_managed_cluster_addons(resource_name, namespace)

Get Managed Cluster Addons obj

Parameters:

resource_name (str) – resource name
namespace (str) – namespace

Returns:

ocp object of managed cluster addons resource

Return type:

ocp_obj

ocs_ci.helpers.helpers.get_memory_leak_median_value()

Function to calculate memory leak Median value by collecting the data for 180 sec and find the median value which will be considered as starting point to evaluate memory leak using “RES” value of ceph-osd daemon i.e. list[7] in code

Returns:: dict of worker nodes and respective median value
Return type:: median_dict (dict)

ocs_ci.helpers.helpers.get_mon_db_size_in_kb(mon_pod_obj)

Get mon db size and returns the size in KB The output of ‘du -sh’ command contains the size of the directory and its path as string e.g. “67M /var/lib/ceph/mon/ceph-c/store.db” The size is extracted by splitting the string with ‘ ‘. The size format for example: 1K, 234M, 2G For uniformity, this test uses KB

Parameters:: mon_pod_obj (obj) – Mon pod resource object
Returns:: Converted Mon db size in KB
Return type:: convert_device_size (int)

ocs_ci.helpers.helpers.get_mon_pdb()

Check for Mon PDB

Returns:: Count of mon allowed disruption min_available_mon (int): Count of minimum mon available max_unavailable_mon (int): Count of maximun mon unavailable
Return type:: disruptions_allowed (int)

ocs_ci.helpers.helpers.get_network_attachment_definitions(nad_name, namespace=None)

Get NetworkAttachmentDefinition obj

Parameters:

nad_name (str) – network_attachment_definition name
namespace (str) – Namespace of the resource

Returns:

network_attachment_definitions object

Return type:

network_attachment_definitions (obj)

ocs_ci.helpers.helpers.get_node_plugin_label(interface)

Identify the csi plugin label based on deployment mode and version

Parameters:: interface (str) – CephFileSystem or CephBlockPool
Returns:: Label of the pod
Return type:: str

ocs_ci.helpers.helpers.get_noobaa_db_credentials_from_secret()

Get credentials details i.e., user and password from noobaa-db secret

Returns:: Username for the db password: Password for the db
Return type:: user_name

ocs_ci.helpers.helpers.get_noobaa_db_size()

Get noobaa db size

Returns:: Noobaa db size
Return type:: str

ocs_ci.helpers.helpers.get_noobaa_db_usage_percent()

Get noobaa db usage percentage

Returns:: Noobaa db usage percentage
Return type:: str

ocs_ci.helpers.helpers.get_noobaa_db_used_space()

Get noobaa db size

Returns:: noobaa_db used space
Return type:: df_out (str)

ocs_ci.helpers.helpers.get_noobaa_metrics_token_from_secret(): Retrieve the JWT needed to authenticate NooBaa metrics queries

ocs_ci.helpers.helpers.get_noobaa_url()

Get the URL of noobaa console

Returns:: url of noobaa console
Return type:: str

ocs_ci.helpers.helpers.get_pods_nodes_logs()

Get logs from all pods and nodes

Returns:: node/pod name as key, logs content as value (string)
Return type:: dict

ocs_ci.helpers.helpers.get_pool_cr(pool_name)

Get the pool CR even if the kind is unknown.

Parameters:: pool_name (str) – The name of the pool to get the CR for.
Returns:: If the resource is found, None otherwise.
Return type:: dict

ocs_ci.helpers.helpers.get_pool_size_factor(pool_name)

Return the raw-storage multiplier for a given Ceph pool.

When data is written to a pool, the pool’s raw size_bytes (as reported by rados df) grows by logical_bytes * factor.

Replicated pool – factor equals the replica count (size). E.g. size=3 → factor=3.
Erasure-Coded pool – factor equals (k + m) / k where k is spec.erasureCoded.dataChunks and m is spec.erasureCoded.codingChunks from the CephBlockPool CR. E.g. k=2, m=1 → factor=1.5.

The factor is derived from the CephBlockPool CR without running any Ceph commands, so no toolbox pod is required.

Parameters:: pool_name (str) – Name of the CephBlockPool resource.
Returns:: Raw-storage multiplier for expected-size calculations.
Return type:: float

ocs_ci.helpers.helpers.get_provision_time(interface, pvc_name, status='start')

Get the starting/ending creation time of a PVC based on provisioner logs

Parameters:

interface (str) – The interface backed the PVC
pvc_name (str / list) – Name of the PVC(s) for creation time the list will be list of pvc objects
status (str) – the status that we want to get - Start / End

Returns:

Time of PVC(s) creation

Return type:

datetime object

ocs_ci.helpers.helpers.get_provisioner_label(interface)

Identify the csi provisioner label based on deployment mode and version

Parameters:: interface (str) – CephFileSystem or CephBlockPool
Returns:: Label of the pod
Return type:: str

ocs_ci.helpers.helpers.get_pv_names()

Get Pv names

Returns:: list of pv names
Return type:: list

ocs_ci.helpers.helpers.get_pv_size(storageclass=None)

Get Pv size from requested storageclass

Parameters:: storageclass (str) – Name of storageclass
Returns:: list of pv’s size
Return type:: list

ocs_ci.helpers.helpers.get_rbd_daemonset_csi_addons_node_object(node)

Gets rdb daemonset CSI addons node data

Parameters:: node (str) – Name of the node
Returns:: CSI addons node object info
Return type:: dict

ocs_ci.helpers.helpers.get_rbd_image_info(rbd_pool, rbd_image_name)

Get RBD image information. (e.g provisioned size, used size, image , )

Parameters:

rbd_pool (str) – RBD metadata pool name (i.e. parameters.pool from the StorageClass, not parameters.dataPool). On EC clusters this is the replicated metadata pool (e.g. replicated-metadata-pool).
rbd_image_name (str) – name of rbd image

Returns:

rbd image information e.g, provisioned size, used size etc.

Return type:

dict

ocs_ci.helpers.helpers.get_rbd_sc_name()

Get the rbd storage class name.

Returns:: The rbd storage class name.
Return type:: str
Raises:: ValueError – If the rbd storage class name hasn’t been found.

ocs_ci.helpers.helpers.get_reclaimspacecronjob_for_pvc(pvc_obj)

Retrieve the ReclaimSpaceCronJob object associated with a given PVC.

Parameters:: pvc_obj (object) – PersistentVolumeClaim (PVC) object.
Returns:: OCP object representing the ReclaimSpaceCronJob associated with the PVC.
Return type:: object
Raises:: ValueError – If the PVC does not have the required annotation for ReclaimSpaceCronJob.

ocs_ci.helpers.helpers.get_rook_ceph_pod_events(pod_name)

Get the rook ceph pod events from the rook ceph pod operator logs

Parameters:: pod_name (str) – The rook ceph pod name to get the events
Returns:: List of all the event lines with the specific pod
Return type:: list

ocs_ci.helpers.helpers.get_rook_ceph_pod_events_by_keyword(pod_name, keyword)

Get the rook ceph pod events with the keyword ‘keyword’ from the rook ceph pod operator logs

Parameters:

pod_name (str) – The rook ceph pod name to get the events
keyword (str) – The keyword to search in the events

Returns:

List of all the event lines with the specific pod that has the keyword ‘keyword’

Return type:

list

ocs_ci.helpers.helpers.get_s3_credentials_from_secret(secret_name)

ocs_ci.helpers.helpers.get_schedule_precedance_value_from_csi_addons_configmap(default='storageclass')

Return the schedule precedence from the ‘csi-addons-config’ ConfigMap.

Delegates to the generic get_csi_addons_config_value() helper, then validates the returned value is ‘storageclass’ or ‘pvc’.

Parameters:: default (str) – Default value to return if ConfigMap doesn’t exist. Defaults to ‘storageclass’.
Returns:: The schedule precedence value (‘storageclass’ or ‘pvc’).
Return type:: str

ocs_ci.helpers.helpers.get_secret_names(namespace=None, resource_name='')

Get secrets names

Parameters:

namespace (str) – The name of the project.
resource_name (str) – The resource name to fetch.

Returns:

secret names

Return type:

dict

ocs_ci.helpers.helpers.get_serviceaccount_obj(sa_name, namespace)

Get serviceaccount obj

Parameters:

sa_name (str) – Service Account name
namespace (str) – The namespace for the serviceaccount creation

Returns:

An OCS instance for the service_account

Return type:

OCS

ocs_ci.helpers.helpers.get_snapshot_content_obj(snap_obj)

Get volume snapshot content of a volume snapshot

Parameters:: snap_obj (OCS) – OCS instance of kind VolumeSnapshot
Returns:: OCS instance of kind VolumeSnapshotContent
Return type:: OCS

ocs_ci.helpers.helpers.get_start_creation_time(interface, pvc_name)

Get the starting creation time of a PVC based on provisioner logs

Parameters:

interface (str) – The interface backed the PVC
pvc_name (str) – Name of the PVC for creation time measurement

Returns:

Start time of PVC creation

Return type:

datetime object

ocs_ci.helpers.helpers.get_start_deletion_time(interface, pv_name)

Get the starting deletion time of a PVC based on provisioner logs

Parameters:

interface (str) – The interface backed the PVC
pvc_name (str) – Name of the PVC for deletion time measurement

Returns:

Start time of PVC deletion

Return type:

datetime object

ocs_ci.helpers.helpers.get_volsync_channel()

Get Volsync Channel

Returns:: volsync channel
Return type:: str

ocs_ci.helpers.helpers.induce_mon_quorum_loss()

Take mon quorum out by deleting /var/lib/ceph/mon directory so that it will start crashing and the quorum is lost

Returns:: List of mon objects mon_pod_running[0] (obj): A mon object which is running ceph_mon_daemon_id (list): List of crashed ceph mon id
Return type:: mon_pod_obj_list (list)

ocs_ci.helpers.helpers.ip_from_subnet_offset(subnet: str, offset: int) → str

Return an IP address from a subnet offset from the network address.

The function takes a subnet in CIDR notation and returns the IP address obtained by adding the given offset to the subnet’s network address.

Parameters:

subnet (str) – Subnet in CIDR notation (e.g. “192.168.252.0/24”).
offset (int) – Number of IP addresses to add to the network address.

Returns:

The resulting IP address as a string.

Return type:

str

Raises:

ValueError – If the subnet is invalid or the resulting IP is outside of the subnet range.

Example

ip_from_subnet_offset(“192.168.252.0/24”, 5) ‘192.168.252.5’ ip_from_subnet_offset(“192.168.252.16/28”, 5) ‘192.168.252.21’

ocs_ci.helpers.helpers.is_pvc_encrypted(pvc_obj)

Check if a PVC is encrypted by examining its StorageClass and PV attributes.

Parameters:: pvc_obj (PVC) – PVC object to check
Returns:: True if PVC is encrypted, False otherwise
Return type:: bool

ocs_ci.helpers.helpers.is_rbd_default_storage_class(sc_name=None)

Check if RDB is a default storageclass for the cluster

Parameters:: custom_sc – custom storageclass name.
Returns:: True if RBD is set as the Default storage class for the cluster, False otherwise.
Return type:: bool

ocs_ci.helpers.helpers.is_volume_present_in_backend(interface, image_uuid, pool_name=None)

Check whether Image/Subvolume is present in the backend.

Parameters:

interface (str) – The interface backed the PVC
image_uuid (str) – Part of VolID which represents corresponding image/subvolume in backend, eg: oc get pv/<volumeName> -o jsonpath='{.spec.csi.volumeHandle}' Output is the CSI generated VolID and looks like: 0001-000c-rook-cluster-0000000000000001-f301898c-a192-11e9-852a-1eeeb6975c91 where image_uuid is f301898c-a192-11e9-852a-1eeeb6975c91
pool_name (str) – Name of the rbd-pool if interface is CephBlockPool

Returns:

True if volume is present and False if volume is not present

Return type:

bool

ocs_ci.helpers.helpers.label_worker_node(node_list, label_key, label_value)

Function to label worker node for running app pods on specific worker nodes.

Parameters:

node_list (list) – List of node name
label_key (str) – Label_key to be added in worker
label_value (str) – Label_value

ocs_ci.helpers.helpers.measure_pv_deletion_time_bulk(interface, pv_name_list, wait_time=60, return_log_times=False)

Measure PV deletion time of bulk PV, based on logs.

Parameters:

interface (str) – The interface backed the PV
pv_name_list (list) – List of PV Names for measuring deletion time
wait_time (int) – Seconds to wait before collecting CSI log
return_log_times (bool) – Determines the return value – if False, dictionary of pv_names with the deletion time is returned; if True – the dictionary of pv_names with the tuple of (srart_deletion_time, end_deletion_time) is returned

Returns:

Dictionary where the pv_names are the keys. The value of the dictionary depend on the: return_log_times argument value and are either the corresponding deletion times (when return_log_times is False) or a tuple of (start_deletion_time, end_deletion_time) as they appear in the logs

Return type:

pv_dict (dict)

ocs_ci.helpers.helpers.measure_pvc_creation_time(interface, pvc_name)

Measure PVC creation time based on logs

Parameters:: interface (str) – The interface backed the PVC pvc_name (str): Name of the PVC for creation time measurement
Returns:: Creation time for the PVC
Return type:: float

ocs_ci.helpers.helpers.measure_pvc_creation_time_bulk(interface, pvc_name_list, wait_time=60)

Measure PVC creation time of bulk PVC based on logs.

Parameters:

interface (str) – The interface backed the PVC
pvc_name_list (list) – List of PVC Names for measuring creation time
wait_time (int) – Seconds to wait before collecting CSI log

Returns:

Dictionary of pvc_name with creation time.

Return type:

pvc_dict (dict)

ocs_ci.helpers.helpers.measure_pvc_deletion_time(interface, pv_name)

Measure PVC deletion time based on logs

Parameters:

interface (str) – The interface backed the PVC
pv_name (str) – Name of the PV for creation time measurement

Returns:

Deletion time for the PVC

Return type:

float

ocs_ci.helpers.helpers.memory_leak_analysis(median_dict)

Function to analyse Memory leak after execution of test case Memory leak is analyzed based on top output “RES” value of ceph-osd daemon, i.e. list[7] in code.

More Detail on Median value: For calculating memory leak require a constant value, which should not be start or end of test, so calculating it by getting memory for 180 sec before TC execution and take a median out of it. Memory value could be different for each nodes, so identify constant value for each node and update in median_dict

Parameters:

median_dict (dict) – dict of worker nodes and respective median value
eg – median_dict = {‘worker_node_1’:102400, ‘worker_node_2’:204800, …}

Usage:

test_case(.., memory_leak_function):
    .....
    median_dict = helpers.get_memory_leak_median_value()
    .....
    TC execution part, memory_leak_fun will capture data
    ....
    helpers.memory_leak_analysis(median_dict)
    ....

ocs_ci.helpers.helpers.modify_deployment_replica_count(deployment_name, replica_count, namespace=None)

Function to modify deployment replica count, i.e to scale up or down deployment

Parameters:

deployment_name (str) – Name of deployment
replica_count (int) – replica count to be changed to
namespace (str) – namespace where the deployment exists

Returns:

True in case if changes are applied. False otherwise

Return type:

bool

ocs_ci.helpers.helpers.modify_deploymentconfig_replica_count(deploymentconfig_name, replica_count, namespace=None)

Function to modify deploymentconfig replica count, i.e to scale up or down deploymentconfig

Parameters:

deploymentcofig_name (str) – Name of deploymentconfig
replica_count (int) – replica count to be changed to
namespace (str) – namespace where the deploymentconfig exists

Returns:

True in case if changes are applied. False otherwise

Return type:

bool

ocs_ci.helpers.helpers.modify_job_parallelism_count(job_name, count, namespace=None)

Function to modify Job instances count,

Parameters:

job_name (str) – Name of job
count (int) – instances count to be changed to
namespace (str) – namespace where the job is running

Returns:

True in case if changes are applied. False otherwise

Return type:

bool

ocs_ci.helpers.helpers.modify_osd_replica_count(resource_name, replica_count)

Function to modify osd replica count to 0 or 1

Parameters:

resource_name (str) – Name of osd i.e, ‘rook-ceph-osd-0-c9c4bc7c-bkf4b’
replica_count (int) – osd replica count to be changed to

Returns:

True in case if changes are applied. False otherwise

Return type:

bool

ocs_ci.helpers.helpers.modify_statefulset_replica_count(statefulset_name, replica_count, namespace=None)

Function to modify statefulset replica count, i.e to scale up or down statefulset

Parameters:

statefulset_namee (str) – Name of statefulset
replica_count (int) – replica count to be changed to

Returns:

True in case if changes are applied. False otherwise

Return type:

bool

ocs_ci.helpers.helpers.mon_pods_running_on_same_node(): Verifies two mons are running on same node

ocs_ci.helpers.helpers.odf_cli_set_log_level(service, log_level, subsystem)

Set the log level for a Ceph service. :param service: The Ceph service name. :type service: str :param log_level: The log level to set. :type log_level: str :param subsystem: The subsystem for which to set the log level. :type subsystem: str

Returns:: The output of the command execution.
Return type:: str

ocs_ci.helpers.helpers.pod_start_time(pod_obj)

Function to measure time taken for container(s) to get into running state by measuring the difference between container’s start time (when container went into running state) and started time (when container was actually started)

Parameters:: pod_obj (obj) – pod object to measure start time
Returns:: Returns the name and start time of container(s) in a pod
Return type:: containers_start_time(dict)

ocs_ci.helpers.helpers.pull_images(image_name)

Function to pull images on all nodes

Parameters:: image_name (str) – Name of the container image to be pulled

Returns: None

ocs_ci.helpers.helpers.recover_mon_quorum(mon_pod_obj_list, mon_pod_running, ceph_mon_daemon_id)

Recover mon quorum back by following procedure mentioned in https://access.redhat.com/solutions/5898541

Parameters:

mon_pod_obj_list (list) – List of mon objects
mon_pod_running (obj) – A mon object which is running
ceph_mon_daemon_id (list) – List of crashed ceph mon id

ocs_ci.helpers.helpers.refresh_oc_login_connection(user=None, password=None)

Function to refresh oc user login Default login using kubeadmin user and password

Parameters:

user (str) – Username to login
password (str) – Password to login

ocs_ci.helpers.helpers.remove_label_from_worker_node(node_list, label_key)

Function to remove label from worker node.

Parameters:

node_list (list) – List of node name
label_key (str) – Label_key to be remove from worker node

ocs_ci.helpers.helpers.remove_ocs_object_from_list(kind, resource_name, object_list)

Given a list of OCS objects, the function removes the object with kind and resource from the list

Parameters:

kind (str) – resource kind like CephBlockPool, pvc.
resource_name (str) – name of the resource.
object_list (array) – Array of OCS objects.

Returns:

Array of OCS objects without removed object.

Return type:

(array)

ocs_ci.helpers.helpers.remove_port_from_url(url)

Remove the port from a URL while preserving the scheme, hostname, and path. :param url: The URL to sanitize. :type url: str

Returns:: The URL without any port information.
Return type:: str

ocs_ci.helpers.helpers.remove_scc_policy(sa_name, namespace)

Removing ServiceAccount from scc anyuid and privileged

Parameters:

sa_name (str) – ServiceAccount name
namespace (str) – The namespace for the scc_policy deletion

ocs_ci.helpers.helpers.remove_toleration()

Remove toleration on storagecluster, subscription and drivers

Returns:: True if all operations succeed, otherwise False.
Return type:: bool

ocs_ci.helpers.helpers.reset_all_osd_pods(): Reset all osd pods

ocs_ci.helpers.helpers.restart_node_if_debug_doesnt_work(worker_node_name)

Check if debug command works on node, and if not, restart the node.

Parameters:: worker_node_name (str) – worker node name

ocs_ci.helpers.helpers.retrieve_cli_binary(cli_type='mcg')

Download the MCG-CLI/ODF-CLI binary and store it locally.

Parameters:: cli_type (str) – choose which bin file you want to download [“odf” -> odf-cli , “mcg” -> mcg-cli]
Raises:: AssertionError – In the case the CLI binary is not executable.

ocs_ci.helpers.helpers.retrieve_default_ingress_crt(): Copy the default ingress certificate from the router-ca secret to the local code runner for usage with boto3.

ocs_ci.helpers.helpers.rsync_kubeconf_to_node(node)

Function to copy kubeconfig to OCP node

Parameters:: node (str) – OCP node to copy kubeconfig if not present

ocs_ci.helpers.helpers.run_cmd_verify_cli_output(cmd=None, expected_output_lst=(), cephtool_cmd=False, ocs_operator_cmd=False, debug_node=None)

Run command and verify its output

Parameters:

cmd (str) – cli command
expected_output_lst (set) – A set of strings that need to be included in the command output.
cephtool_cmd (bool) – command on ceph-tool pod
ocs_operator_cmd (bool) – command on ocs-operator pod
debug_node (str) – name of node

Returns:

True of all strings are included in the command output, False otherwise

Return type:

bool

ocs_ci.helpers.helpers.run_io_with_rados_bench(**kw)

A task for radosbench. Runs radosbench command on specified pod . If parameters are not provided task assumes few default parameters.This task runs command in synchronous fashion.

Parameters:

kw (dict) –

a dictionary of various radosbench parameters. ex:

pool_name:pool
pg_num:number of pgs for pool
op: type of operation {read, write}
cleanup: True OR False

Returns:

return value of radosbench command

Return type:

ret

ocs_ci.helpers.helpers.scale_nb_resources(replica=1)

Function scales noobaa resources

Parameters:: replica (int) – Replica count

ocs_ci.helpers.helpers.select_unique_pvcs(pvcs)

Get the PVCs with unique access mode and volume mode combination.

Parameters:: pvcs (list) – List of PVC objects
Returns:: List of selected PVC objects
Return type:: list

ocs_ci.helpers.helpers.set_configmap_log_level_csi_sidecar(value): Set CSI_SIDECAR log level on configmap of rook-ceph-operator :param value: type of log :type value: int

ocs_ci.helpers.helpers.set_configmap_log_level_rook_ceph_operator(value)

Set ROOK_LOG_LEVEL on configmap of rook-ceph-operator

Parameters:: value (str) – type of log

ocs_ci.helpers.helpers.set_image_lookup(image_name)

Function to enable lookup, which allows reference to the image stream tag in the image field of the object. Example:

$ oc set image-lookup mysql
$ oc run mysql --image=mysql

Parameters:: image_name (str) – Name of the image stream to pull the image locally
Returns:: output of set image-lookup command
Return type:: str

ocs_ci.helpers.helpers.set_rook_log_level(): Set the rook log level

ocs_ci.helpers.helpers.set_schedule_precedence(precedence)

Set the schedule-precedence key in the ‘csi-addons-config’ ConfigMap and restart the CSI Addons controller manager.

Delegates to the generic update_csi_addons_config() helper.

Parameters:: precedence (str) – Must be ‘storageclass’ or ‘pvc’.
Raises:: ValueError – If precedence is not ‘storageclass’ or ‘pvc’.

ocs_ci.helpers.helpers.setup_pod_directories(pod_obj, dir_names)

Creates directories on the specified pod. Directories created under the respective test name directory.

Parameters:

pod_obj – A pod object on which to create directories
dir_names – A list of directories names to create.

Returns:

A list of all the full paths of the created directories

Return type:

list

ocs_ci.helpers.helpers.storagecluster_independent_check()

Check whether the storagecluster is running in independent mode by checking the value of spec.externalStorage.enable

Returns:: True if storagecluster is running on external mode False otherwise
Return type:: bool

ocs_ci.helpers.helpers.unfence_node(node_name, delete=False)

Un-fence node

Parameters:

node_name (str) – Name of the node
delete (bool) – If True, delete the network fence object

ocs_ci.helpers.helpers.update_volsync_channel(): Update Volsync Channel.

ocs_ci.helpers.helpers.upgrade_multus_holder_design(): Upgrade multus holder design from ODF4.15 to ODF4.16

ocs_ci.helpers.helpers.validate_cephfilesystem(fs_name, namespace=None)

Verify CephFileSystem exists at Ceph and OCP

Parameters:

fs_name (str) – The name of the Ceph FileSystem

Returns:

True if CephFileSystem is created at Ceph and OCP side else: will return False with valid msg i.e Failure cause

Return type:

bool

ocs_ci.helpers.helpers.validate_pod_oomkilled(pod_name, namespace=None, container=None)

Validate pod oomkilled message are found on log

Parameters:

pod_name (str) – Name of the pod
namespace (str) – Namespace of the pod
container (str) – Name of the container

Returns:

True if oomkill messages are not found on log.: False Otherwise.

Return type:

bool

Raises:

Assertion if failed to fetch logs –

ocs_ci.helpers.helpers.validate_pods_are_running_and_not_restarted(pod_name, pod_restart_count, namespace)

Validate given pod is in running state and not restarted or re-spinned

Parameters:

pod_name (str) – Name of the pod
pod_restart_count (int) – Restart count of pod
namespace (str) – Namespace of the pod

Returns:

True if pod is in running state and restart: count matches the previous one

Return type:

bool

ocs_ci.helpers.helpers.validate_pv_delete(pv_name)

validates if pv is deleted after pvc deletion

Parameters:: pv_name (str) – pv from pvc to validates
Returns:: True if deletion is successful
Return type:: bool
Raises:: AssertionError – If pv is not deleted

ocs_ci.helpers.helpers.validate_scc_policy(sa_name, namespace, scc_name='privileged')

Validate serviceaccount is added to scc of privileged

Parameters:

sa_name (str) – Service Account name
namespace (str) – The namespace for the serviceaccount creation
scc_name (str) – SCC name

Returns:

True if sc_name is present in scc of privileged else False

Return type:

bool

ocs_ci.helpers.helpers.verify_block_pool_exists(pool_name)

Verify if a Ceph block pool exist

Parameters:: pool_name (str) – The name of the Ceph block pool
Returns:: True if the Ceph block pool exists, False otherwise
Return type:: bool

ocs_ci.helpers.helpers.verify_csi_holder_pods_do_not_exist()

Verify csi holder pods do not exist

Raises:: TimeoutExpiredError – if csi-holder pod exist raise Exception

ocs_ci.helpers.helpers.verify_log_exist_in_pods_logs(pod_names, expected_log, container=None, namespace=None, all_containers_flag=True, since=None)

Verify log exist in pods logs.

Parameters:

pod_names (list) – Name of the pod
expected_log (str) – the expected logs in “oc logs” command
container (str) – Name of the container
namespace (str) – Namespace of the pod
all_containers_flag (bool) – fetch logs from all containers of the resource
since (str) – only return logs newer than a relative duration like 5s, 2m, or 3h.

Returns:

return True if log exist otherwise False

Return type:

bool

ocs_ci.helpers.helpers.verify_nb_db_psql_version(check_image_name_version=True)

Verify that the NooBaa DB PostgreSQL version matches the expectation that is derived from the NooBaa CR.

Parameters:

check_image_name_version (bool) – If True, also check that the version from the name of the image in the NooBaa DB Statefulset matches the one queried from the DB.

Raises:

AssertionError – If the NooBaa DB PostgreSQL version doesn’t match the NooBaa CR expectation.
UnexpectedBehaviour – If the parsing or extraction of the versions fails due to changes in the CRs.

ocs_ci.helpers.helpers.verify_pdb_mon(disruptions_allowed, max_unavailable_mon)

Compare between the PDB status and the expected PDB status

Parameters:

disruptions_allowed (int) – the expected number of disruptions_allowed
max_unavailable_mon (int) – the expected number of max_unavailable_mon

Returns:

True if the expected pdb state equal to actual pdb state, False otherwise

Return type:

bool

ocs_ci.helpers.helpers.verify_performance_profile_change(perf_profile)

Verify that newly applied performance profile got updated in storage cluster

Parameters:: perf_profile (str) – Applied performance profile
Returns:: True in case performance profile is updated, False otherwise
Return type:: bool

ocs_ci.helpers.helpers.verify_pod_pattern_does_not_exist(pattern, namespace)

Verify csi-holder pods do not exist

Parameters:

pattern (str) – the pattern of pod
namespace (str) – the namespace of pod

Returns:

if pod with pattern exist return False otherwise return True

Return type:

bool

ocs_ci.helpers.helpers.verify_pv_mounted_on_node(node_pv_dict)

Check if mount point of a PV exists on a node

Parameters:

node_pv_dict (dict) – Node to PV list mapping eg: {‘node1’: [‘pv1’, ‘pv2’, ‘pv3’], ‘node2’: [‘pv4’, ‘pv5’]}

Returns:

Node to existing PV list mapping: eg: {‘node1’: [‘pv1’, ‘pv3’], ‘node2’: [‘pv5’]}

Return type:

dict

ocs_ci.helpers.helpers.verify_pvc_size(pod_obj, expected_size)

Verify PVC size is as expected or not.

Parameters:

pod_obj – Pod Object
expected_size – Expected size of PVC

Returns:

True if expected size is matched with the PVC attached to pod. else False

Return type:

bool

ocs_ci.helpers.helpers.verify_quota_resource_exist(quota_name)

Verify quota resource exist

Parameters:: quota_name (str) – The name of quota
Returns:: return True if quota_name exist in list, otherwise False
Return type:: bool

ocs_ci.helpers.helpers.verify_reclaimspacecronjob_suspend_state_for_pvc(pvc_obj)

Verify the suspend state of the ReclaimSpaceCronJob associated with the given PVC.

Parameters:: pvc_obj (object) – PersistentVolumeClaim (PVC) object.
Returns:: True if the suspend state is True and the state annotation is ‘unmanaged’, False otherwise.
Return type:: bool

ocs_ci.helpers.helpers.verify_rook_ceph_crashcollector_pods_where_rook_ceph_pods_are_running(timeout=90)

Verify rook-ceph-crashcollector pods running on worker nodes where rook-ceph pods are running.

Parameters:

timeout (int) – time to wait for verifying

Returns:

True if rook-ceph-crashcollector pods running on worker nodes: where rook-ceph pods are running in the given timeout. False otherwise.

Return type:

bool

ocs_ci.helpers.helpers.verify_socket_on_node(node_name, host_path, socket_name)

Verify the existence of socket at host path on node.

Parameters:

node_name (str) – The name of specific node
host_path (str) – The host path where socket exist
socket_name (str) – The name of socket file

Returns:

True if the socket file exist at host path on given node.

Return type:

bool

ocs_ci.helpers.helpers.verify_storagecluster_nodetopology()

Verify only nodes with OCS label in storagecluster under nodeTopologies block

Returns:: return True if storagecluster contain only nodes with OCS label
Return type:: bool

ocs_ci.helpers.helpers.verify_substrings_in_string(output_string, expected_strings)

Verify all expected substrings are present in the output string

Parameters:

output_string (str) – String to check
expected_strings (list) – Expected substrings

Returns:

True if all expected strings found, False otherwise

Return type:

bool

ocs_ci.helpers.helpers.verify_volume_deleted_in_backend(interface, image_uuid, pool_name=None, timeout=180)

Ensure that Image/Subvolume is deleted in the backend.

Parameters:

interface (str) – The interface backed the PVC
image_uuid (str) – Part of VolID which represents corresponding image/subvolume in backend, eg: oc get pv/<volumeName> -o jsonpath='{.spec.csi.volumeHandle}' Output is the CSI generated VolID and looks like: 0001-000c-rook-cluster-0000000000000001-f301898c-a192-11e9-852a-1eeeb6975c91 where image_uuid is f301898c-a192-11e9-852a-1eeeb6975c91
pool_name (str) – Name of the rbd-pool if interface is CephBlockPool
timeout (int) – Wait time for the volume to be deleted.

Returns:

True if volume is deleted before timeout.: False if volume is not deleted.

Return type:

bool

ocs_ci.helpers.helpers.wait_for_ct_pod_recovery()

In case the of node failures scenarios, in which the selected node is running the ceph tools pod, we’ll want to wait for the pod recovery

Returns:: True in case the ceph tools pod was recovered, False otherwise
Return type:: bool

ocs_ci.helpers.helpers.wait_for_osds_down(osd_ids: list[str], timeout: int = 300, sleep: int = 10) → None

Wait for OSDs to be marked as ‘down’ in Ceph using polling

Parameters:

osd_ids (list[str]) – List of OSD IDs to wait for
timeout (int) – Timeout in seconds (default: 300)
sleep (int) – Sleep interval between checks (default: 10)

Raises:

TimeoutExpiredError – If OSDs don’t go down within timeout

ocs_ci.helpers.helpers.wait_for_pv_delete(pv_objs, timeout=180)

Wait for PVs to delete. Delete PVs having ReclaimPolicy ‘Retain’

Parameters:: pv_objs (list) – OCS instances of kind PersistentVolume

ocs_ci.helpers.helpers.wait_for_quota_usage_update(clusterresourcequota_obj, quota_name, quota_key, expected_strings, operation_description, timeout=120)

Wait for ClusterResourceQuota usage to update and verify expected strings

Parameters:

clusterresourcequota_obj (obj) – ClusterResourceQuota OCP object
quota_name (str) – Name of the quota resource
quota_key (str) – Quota key to check (e.g., ‘requests.storage’)
expected_strings (list) – Strings to verify (e.g., [‘8Gi’, ‘7Gi’])
operation_description (str) – Operation description for logging
timeout (int) – Timeout in seconds (default: 120)

Raises:

TimeoutExpiredError – If quota usage doesn’t update within timeout

ocs_ci.helpers.helpers.wait_for_reclaim_space_cronjob(reclaim_space_cron_job, schedule)

Wait for reclaim space cronjbo

Parameters:

reclaim_space_cron_job (obj) – The reclaim space cron job
schedule (str) – Reclaim space cron job schedule

Raises:

UnexpectedBehaviour – In case reclaim space cron job doesn’t reach the desired state

ocs_ci.helpers.helpers.wait_for_reclaim_space_job(reclaim_space_job)

Wait for reclaim space cronjbo

Parameters:: reclaim_space_job (obj) – The reclaim space job
Raises:: UnexpectedBehaviour – In case reclaim space job doesn’t reach the Succeeded state

ocs_ci.helpers.helpers.wait_for_resource_count_change(func_to_use, previous_num, namespace, change_type='increase', min_difference=1, timeout=20, interval=2, **func_kwargs)

Wait for a change in total count of PVC or pod

Parameters:

func_to_use (function) – Function to be used to fetch resource info Supported functions: pod.get_all_pvcs(), pod.get_all_pods()
previous_num (int) – Previous number of pods/PVCs for comparison
namespace (str) – Name of the namespace
change_type (str) – Type of change to check. Accepted values are ‘increase’ and ‘decrease’. Default is ‘increase’.
min_difference (int) – Minimum required difference in PVC/pod count
timeout (int) – Maximum wait time in seconds
interval (int) – Time in seconds to wait between consecutive checks

Returns:

True if difference in count is greater than or equal to: ’min_difference’. False in case of timeout.

ocs_ci.helpers.helpers.wait_for_resource_state(resource, state, timeout=60)

Wait for a resource to get to a given status

Parameters:

resource (OCS obj) – The resource object
state (str) – The status to wait for
timeout (int) – Time in seconds to wait

Raises:

ResourceWrongStatusException – In case the resource hasn’t reached the desired state

ocs_ci.helpers.helpers.wait_for_rook_ceph_pod_status(pod_obj, desired_status, timeout=420)

Wait for the rook ceph pod to reach the desired status. If the pod didn’t reach the desired status, check if the reason is that the pod is not found. If this is the case, check in the rook ceph pod operator logs to see if the pod reached the desired status.

Parameters:

pod_obj (ocs_ci.ocs.resources.pod.Pod) – The rook ceph pod object
desired_status (str) – The desired status of the pod to wait for
timeout (int) – time to wait for the pod to reach the desired status

Returns:

True if the rook ceph pod to reach the desired status. False, otherwise

Return type:

bool

ocs_ci.helpers.helpers.wait_for_volume_detachment(pvc_objs, timeout=180)

Wait until the volumes are fully detached from all nodes by checking the VolumeAttachment resources. This makes sure the volumes are safely removed before deleting.

Parameters:

pvc_objs (list) – List of PVC objects to check for detachment
timeout (int) – Timeout in seconds to wait for detachment (default: 180)

Returns:

True if all volumes are detached, False otherwise

Return type:

bool

ocs_ci.helpers.keyrotation_helper module

class ocs_ci.helpers.keyrotation_helper.KeyRotation

Bases: object

Handles key rotation operations for a storage cluster.

disable_keyrotation()

Disables key rotation for the storage cluster.

Returns:: True if key rotation is disabled, False otherwise.
Return type:: bool

enable_keyrotation()

Enables key rotation for the storage cluster.

Returns:: True if key rotation is enabled, False otherwise.
Return type:: bool

get_keyrotation_schedule()

Retrieves the current key rotation schedule for the storage cluster.

Returns:: The key rotation schedule.
Return type:: str

is_keyrotation_enable()

Checks if key rotation is enabled for the storage cluster.

Returns:: True if key rotation is enabled, False otherwise.
Return type:: bool

set_keyrotation_defaults(): Setting Keyrotation Defaults on the cluster.

set_keyrotation_schedule(schedule)

Sets the key rotation schedule for the storage cluster.

Parameters:: schedule (str) – The new key rotation schedule.

class ocs_ci.helpers.keyrotation_helper.NoobaaKeyrotation

Bases: KeyRotation

Extends KeyRotation class to handle key rotation operations for Noobaa.

get_noobaa_backend_secret(kms_deployment=False)

Retrieves the backend secret for Noobaa.

kms_deployment: Boolean: Sets as False, if True it will check NOOBAA_BACKEND_SECRET in vault

Returns:: containing the Noobaa backend root key and secret.
Return type:: tuple (str, str)
Raises:: ValueError – If failed to retrieve the backend secret.

get_noobaa_keyrotation_schedule()

Retrieves the current key rotation schedule for Noobaa.

Returns:: The key rotation schedule for Noobaa.
Return type:: str
Raises:: ValueError – If the key rotation schedule is not found or is invalid.

get_noobaa_volume_secret()

Retrieves the volume secret for Noobaa.

Returns:: containing the Noobaa volume root key and secret.
Return type:: tuple (str, str)
Raises:: ValueError – If failed to retrieve the volume secret.

is_noobaa_keyrotation_enable()

Checks if key rotation is enabled for Noobaa.

Returns:: True if key rotation is enabled for Noobaa, False otherwise.
Return type:: bool

class ocs_ci.helpers.keyrotation_helper.OSDKeyrotation

Bases: KeyRotation

Extends KeyRotation class to handle key rotation operations for Rook.

enable_osd_keyrotatio()

Enable OSD keyrotation in storagecluster Spec.

Returns:: True if keyrotation is Enabled otherwise False
Return type:: bool

get_osd_dm_crypt(device)

Retrieves the dmcrypt key for OSD.

Parameters:: device (str) – The OSD device name.
Returns:: The dmcrypt key for the specified OSD device.
Return type:: str

get_osd_keyrotation_schedule()

Retrieves the key rotation schedule for OSD.

Returns:: The key rotation schedule for OSD.
Return type:: str

is_osd_keyrotation_enabled()

Checks if key rotation is enabled for OSD.

Returns:: True if key rotation is enabled for OSD, False otherwise.
Return type:: bool

verify_keyrotation(old_keys, tries=10, delay=20)

Verify Keyrotation is suceeded for all OSD devices.

Parameters:: old_keys (dict) – osd devices and their keys.
Returns:: True if all OSD keyrotation is happend, orherwise False.
Return type:: bool

verify_osd_keyrotation_for_kms(tries=10, delay=10)

Verify OSD KeyRotation for Vault KMS

Returns:: return True If KeyRotation is sucessfull otherwise False.
Return type:: bool

class ocs_ci.helpers.keyrotation_helper.PVKeyrotation(sc_obj)

Bases: KeyRotation

annotate_storageclass_key_rotation(schedule='@weekly'): Annotate Storageclass To enable keyrotation for encrypted PV

change_pvc_keyrotation_cronjob_state(pvc_objs, disable=True)

Modify the key rotation state of PVCs by annotating and patching their associated cronjobs.

Parameters:

pvc_objs (list) – List of PVC objects to modify.
disable (bool) – If True, disables the key rotation. If False, enables it. Defaults to True.

Returns:

True if the operation succeeds.

Return type:

bool

compare_keys(device_handle, old_key)

Compares the current key with the rotated key.

Parameters:

device_handle (str) – The handle or identifier for the device.
old_key (str) – The current key before rotation.

Returns:

True if the key has rotated successfully.

Return type:

bool

Raises:

UnexpectedBehaviour – If the keys have not rotated.

get_keyrotation_cronjob_for_pvc(pvc_obj)

Retrieves the key rotation CronJob associated with a PVC.

Parameters:: pvc_obj (object) – The PVC object for which to retrieve the CronJob.
Returns:: The CronJob object associated with the PVC.
Return type:: object
Raises:: ValueError – If the PVC lacks the key rotation CronJob annotation and no cronjob can be found by naming convention.

get_pvc_keys_data(pvc_objs)

Retrieves key data for PVCs.

Returns:

Dictionary mapping PVC names to their device handles and vault keys.: Returns None for vault_key if the key doesn’t exist yet.

Return type:

dict

get_pvc_kms_id(pvc_obj)

Get the encryption KMS ID for a PVC from its PV volumeAttributes.

Parameters:: pvc_obj (object) – The PVC object
Returns:: The encryption KMS ID, or None if not found
Return type:: str

reset_keyrotation_baseline(): Resets the baseline key data for key rotation verification. This should be called after re-enabling key rotation to ensure the baseline is captured after re-enabling, not before.

set_keyrotation_state_by_annotation(enable: bool): Enables or disables key rotation by annotating the StorageClass.

set_keyrotation_state_by_rbac_user(pvc_obj, suspend_state=True): Updates key rotation CronJob state for a PVC.

wait_for_keyrotation_cronjobs_deletion(pvc_objs, timeout=300, interval=10)

Wait for key rotation cronjobs to be deleted for all PVCs after disabling.

Parameters:

pvc_objs (list) – List of PVC objects to check.
timeout (int) – Maximum time to wait in seconds (default: 300).
interval (int) – Time between checks in seconds (default: 10).

Returns:

True if all cronjobs are deleted.

Return type:

bool

Raises:

UnexpectedBehaviour – If cronjobs are not deleted within timeout.

wait_for_keyrotation_cronjobs_recreation(pvc_objs)

Wait for key rotation cronjobs to be recreated for all PVCs after re-enabling.

Parameters:: pvc_objs (list) – List of PVC objects to check.
Returns:: True if all cronjobs are recreated and active.
Return type:: bool
Raises:: UnexpectedBehaviour – If cronjobs are not recreated within timeout.

wait_till_all_pv_keyrotation_on_vault_kms(pvc_objs)

Waits for all PVC keys to be rotated in the Vault KMS.

This method: 1. On first call: Captures baseline keys (initializes all_pvc_key_data) 2. On subsequent retries: Compares current keys with baseline 3. Raises UnexpectedBehaviour if keys haven’t rotated yet (triggers retry) 4. Returns True when all keys have been rotated

Note: If baseline keys don’t exist yet (None), waits for them to appear before considering rotation complete.

wait_till_keyrotation(device_handle)

Waits until the key rotation occurs for a given device handle.

Parameters:: device_handle (str) – The handle or identifier for the device whose key rotation is to be checked.
Returns:: True if the key rotation is successful, otherwise False.
Return type:: bool

ocs_ci.helpers.keyrotation_helper.compare_noobaa_old_keys_with_new_keys(noobaa_keyrotation, old_noobaa_backend_key, old_noobaa_volume_key): Compare noobaa old keys with new keys. :param noobaa_keyrotation: obj: NoobaaKeyrotation object :param old_noobaa_backend_key: str: old noobaa backend key :param old_noobaa_volume_key: str: old noobaa_volume_key

ocs_ci.helpers.keyrotation_helper.validate_key_rotation_schedules(schedule)

Validate key rotation schedules across different components.

Parameters:: schedule (str) – The expected key rotation schedule.
Raises:: ValueError – If the schedule does not match in any of the components.

ocs_ci.helpers.keyrotation_helper.verify_new_key_after_rotation(tries, delays): This function records existing keys for OSD, Noobaa volume and backend and compare with the new keys generated on given schedule.

ocs_ci.helpers.longevity_helpers module

ocs_ci.helpers.longevity_helpers.create_restore_verify_snapshots(multi_snapshot_factory, snapshot_restore_factory, pod_factory, pvc_objs, namespace, file_name)

Creates snapshots from each PVC in the provided list of PVCs, Restores new PVCs out of the created snapshots and Verifies data integrity by checking the existence and md5sum of file in the restored PVC.

Parameters:

multi_snapshot_factory – Fixture to create a VolumeSnapshot of each PVC in the provided list of PVCs.
snapshot_restore_factory – Fixture to create a new PVCs out of the VolumeSnapshot provided.
pod_factory – Fixture to create new PODs.
pvc_objs (list) – List of PVC objects for which snapshots are to be created.
namespace (str) – Namespace in which the PVCs are created.
file_name (str) – Name of the file on which FIO is performed.

Returns:

A tuple of size 2 containing a list of restored PVC objects and a list of the pods attached to the: restored PVCs, respectively.

Return type:

tuple

ocs_ci.helpers.longevity_helpers.expand_verify_pvcs(pvc_objs, pod_objs, pvc_size_new, file_name, fio_size)

Expands size of each PVC in the provided list of PVCs, Verifies data integrity by checking the existence and md5sum of file in the expanded PVC and Runs FIO on expanded PVCs and verifies results.

Parameters:

pvc_objs (list) – List of PVC objects which are to be expanded.
pod_objs (list) – List of POD objects attached to the PVCs.
pvc_size_new (int) – Size of the expanded PVC in GB.
file_name (str) – Name of the file on which FIO is performed.
fio_size (int) – Size in MB of FIO.

ocs_ci.helpers.longevity_helpers.measure_pod_creation_time(namespace, num_of_pods)

Measures and Logs the POD Creation Time of all the PODs.

Parameters:

namespace (str) – Namespace in which the PODs are created.
num_of_pods (int) – Number of PODs created.

Logs:: POD Creation Time of all the PODs.

ocs_ci.helpers.longevity_helpers.measure_pod_to_pvc_attach_time(pod_objs)

Measures and Logs Attach Time of all PODs.

Parameters:: pod_objs (list) – List of POD objects for which we have to measure the time.

Logs:: Attach time of all PODs, as well as the average time.

ocs_ci.helpers.longevity_helpers.measure_pvc_creation_time(interface, pvc_objs, start_time)

Measures and Logs PVC Creation Time of all PVCs.

Parameters:

interface (str) – an interface (RBD or CephFS) to run on.
pvc_objs (list) – List of PVC objects for which we have to measure the time.
start_time (str) – Formatted time from which and on to search the relevant logs.

Logs:: PVC Creation Time of all the PVCs.

ocs_ci.helpers.longevity_helpers.measure_pvc_deletion_time(interface, pvc_objs)

Measures and Logs PVC Deletion Time of all PVCs.

Parameters:

interface (str) – an interface (RBD or CephFS) to run on.
pvc_objs (list) – List of PVC objects for which we have to measure the time.

Logs:: PVC Deletion Time of all the PVCs.

ocs_ci.helpers.longevity_helpers.write_empty_files_to_bucket(mcg_obj, awscli_pod_session, bucket_name, test_directory_setup)

Write empty files to bucket and verify if they are created.

Parameters:

mcg_obj (MCG) – An MCG object containing the MCG S3 connection credentials
awscli_pod_session – Fixture to create a new AWSCLI pod for relaying commands.
bucket_name (str) – Name of the bucket on which files are to be written.
test_directory_setup – Fixture to setup test DIRs.

Raises:

UnexpectedBehaviour – Raises an exception if files are not created.

Returns:

A set of names of all bucket objects.

Return type:

Set

ocs_ci.helpers.managed_services module

Managed Services related functionalities

ocs_ci.helpers.managed_services.get_all_storageclassclaims(namespace=None)

Get all storageclassclaims/storageclaims <storageclassclaim changed to storageclaim from ODF 4.16 >

Returns:: OCS objects of kind Storageclassclaim/storageclaim
Return type:: List

ocs_ci.helpers.managed_services.get_ocs_osd_deployer_version()

Get OCS OSD deployer version from CSV

Returns:: OCS OSD deployer version
Return type:: Version

ocs_ci.helpers.managed_services.get_storageclassclaims_of_storageclient(storageclient_name)

Get all storageclassclaims associated with a storageclient

Parameters:: storageclient_name (str) – Name of the storageclient
Returns:: OCS objects of kind Storageclassclaim
Return type:: List

ocs_ci.helpers.managed_services.get_used_capacity(msg)

Verify OSD percent used capacity greate than ceph_full_ratio

Parameters:: msg (str) – message to be logged
Returns:: The percentage of the used capacity in the cluster
Return type:: float

ocs_ci.helpers.managed_services.verify_client_operator_security()

Check ocs-client-operator-controller-manager permissions

Verify runAsUser is not 0
Verify SecurityContext.allowPrivilegeEscalation is set to false
Verify SecurityContext.capabilities.drop contains ALL

ocs_ci.helpers.managed_services.verify_faas_cluster_secrets(): Verify the secrets present in FaaS cluster

ocs_ci.helpers.managed_services.verify_faas_consumer_resources()

Verify resources specific to FaaS consumer

Verify CSV phase
Verify client endpoint
Check that there’s no storagecluster

ocs_ci.helpers.managed_services.verify_faas_provider_resources()

Verify resources specific to FaaS provider cluster

Verify CSV phase
Verify ocs-provider-server pod is Running
Verify ocs-metrics-exporter pod is Running
Verify that Cephcluster is Ready and hostNetworking is True
Verify that the security groups are set up correctly
Check the presence of catalogsource and its state
Check the presence of subscription and its health
Check that mon PVCs have gp3-csi storageclass
Check managedFusionOffering release, usableCapacityInTiB and onboardingValidationKey
Verify the version of Prometheus
Verify aws volumes
Verify configmaps

ocs_ci.helpers.managed_services.verify_faas_provider_storagecluster(sc_data)

Verify provider storagecluster

allowRemoteStorageConsumers: true (for ODF versions lesser than 4.19)
hostNetwork: true
matchExpressions:
key: node-role.kubernetes.io/worker operator: Exists key: node-role.kubernetes.io/infra operator: DoesNotExist
storageProviderEndpoint
annotations:
uninstall.ocs.openshift.io/cleanup-policy: delete uninstall.ocs.openshift.io/mode: graceful
Check the storagecluster resources limits and requests are valid
Verify the Faas provider storagecluster storages

Parameters:: sc_data (dict) – storagecluster data dictionary

ocs_ci.helpers.managed_services.verify_faas_provider_storagecluster_storages(sc_data)

Verify the Faas provider storagecluster storages

Check the storagecluster backingStorageClasses.
Check that the default ocs storage class is exist in the backingStorageClasses.
Check that the type of the default ocs storage class is “gp3”.
Check the defaultStorageProfile value
Check that the default storage profile is found in the StorageProfiles.
Check that the values in the storage profile are correct

Parameters:: sc_data (dict) – The storagecluster data

ocs_ci.helpers.managed_services.verify_faas_resources(): Verify the presence and status of resources in FaaS clusters

ocs_ci.helpers.managed_services.verify_osd_distribution_on_provider(): Verify the OSD distribution on the provider cluster

ocs_ci.helpers.managed_services.verify_osd_used_capacity_greater_than_expected(expected_used_capacity)

Verify OSD percent used capacity greater than ceph_full_ratio

Parameters:: expected_used_capacity (float) – expected used capacity
Returns:: True if used_capacity greater than expected_used_capacity, False otherwise
Return type:: bool

ocs_ci.helpers.managed_services.verify_pods_in_managed_fusion_namespace(): Verify the status of pods in the namespace managed-fusion

ocs_ci.helpers.managed_services.verify_provider_aws_volumes(): Verify provider AWS volumes: 1. Volumes for OSD have size 4096 2. Volumes for OSD have IOPS 12000 3. Namespace should be fusion-storage

ocs_ci.helpers.managed_services.verify_provider_topology()

Verify topology in a Managed Services provider cluster

Verify replica count
Verify total size
Verify OSD size
Verify worker node instance type
Verify worker node instance count
Verify OSD count
Verify OSD CPU and memory

ocs_ci.helpers.managed_services.verify_storageclient(storageclient_name=None, namespace=None, provider_name=None, verify_sc=True)

Verify status, values and resources related to a storageclient

Parameters:

storageclient_name (str) – Name of the storageclient to be verified. If the name is not given, it will be assumed that only one storageclient is present in the cluster.
namespace (str) – Namespace where the storageclient is present. Default value will be taken from ENV_DATA[“cluster_namespace”]
provider_name (str) – Name of the provider cluster to which the storageclient is connected.
verify_sc (bool) – True to verify the storageclassclaims and storageclasses associated with the storageclient.

ocs_ci.helpers.managed_services.verify_storageclient_storageclass_claims(storageclient)

Verify the status of storageclassclaims and the presence of the storageclass associated with the storageclient

Parameters:: storageclient_name (str) – Name of the storageclient

ocs_ci.helpers.mcg_stress_helper module

ocs_ci.helpers.mcg_stress_helper.delete_objects_in_batches(bucket, batch_size)

Delete objects from the bucket in batches

Parameters:

bucket (tuple) – Tuple consisting of backend storage type and bucket object
batch_size (int) – Number of objects to delete at a time

ocs_ci.helpers.mcg_stress_helper.delete_objs_from_bucket(pod_obj, bucket, prev_iteration, event=None, multiplier=1)

Delete all the objects from a bucket

Parameters:

pod_obj (Pod) – Noobaa stress CLI pod object
bucket (Tuple) – Tuple consisting of backend storage type and bucket object
prev_iteration (int) – Iteration number or prefix from where should delete objects
event (threading.Event()) – Event object to signal the execution completion

ocs_ci.helpers.mcg_stress_helper.download_objs_from_bucket(pod_obj, bucket, target_dir, prev_iteration, event=None, multiplier=1)

Download objects from a bucket back to local directory

Parameters:

pod_obj (Pod) – Noobaa stress CLI pod object
bucket (Tuple) – Tuple consisting of backend storage type and bucket object
target_dir (str) – Target directory to download objects
prev_iteration (int) – Iteration number or prefix from where should download objects
event (threading.Event()) – Event object to signal the execution completion

ocs_ci.helpers.mcg_stress_helper.get_mcg_obj(bucket): Get MCG object based on the bucket type

ocs_ci.helpers.mcg_stress_helper.induce_noobaa_failures(nodes, delay=300)

Induce Noobaa specific failures with specific delay between the each type of failures

Parameters:

nodes (PlatformNodes) – PlatformNodes instance
delay (int) – Time delay between the each failure

ocs_ci.helpers.mcg_stress_helper.list_objs_from_bucket(bucket, prev_iteration, event=None)

List objects from bucket

Parameters:

bucket (Tuple) – Tuple consisting of backend storage type and bucket object
prev_iteration (int) – Iteration number or prefix from where should list objects
event (threading.Event()) – Event object to signal the execution completion

ocs_ci.helpers.mcg_stress_helper.run_background_cluster_checks(scale_noobaa_db_pv, event=None, threading_lock=None)

Run background checks to verify noobaa health and cluster health overall

Check Noobaa Health

Check Ceph Health

Check Noobaa db usage

Check for any alerts

Memory and CPU utilization

ocs_ci.helpers.mcg_stress_helper.run_noobaa_metadata_intense_ops(mcg_obj, pod_obj, bucket_factory, bucket, prev_iteration, event=None, multiplier=1)

Perfrom metdata intense operations to stress Noobaa

Parameters:

mcg_obj (MCG) – MCG object
pod_obj (Pod) – Noobaa stress CLI pod object
bucket_factory (fixture) – Pytest fixture for creating bucket
bucket (tuple) – Tuple consisting of backend storage type and bucket object
prev_iteration (int) – Iteration number or prefix from where should delete objects
event (threading.Event()) – Event object to signal the execution completion

ocs_ci.helpers.mcg_stress_helper.sync_object_directory_with_retry(pod_obj, src, target, s3_obj=None, timeout=None): Wrapper function that will retry sync_object_directory :param pod_obj: Pod object representing stress-cli pod :type pod_obj: Pod :param src: Source directory :type src: str :param target: fully qualified target bucket path :type target: str :param s3_obj: MCG object :type s3_obj: MCG :param timeout: Timeout for s3 sync command :type timeout: int

ocs_ci.helpers.mcg_stress_helper.upload_objs_to_buckets(mcg_obj, pod_obj, buckets, current_iteration, event=None, multiplier=1)

This will upload objects present in the stress-cli pod to the buckets provided concurrently

Parameters:

mcg_obj (MCG) – MCG object
pod_obj (Pod) – Pod object
buckets (Dict) – Map of bucket type and bucket object
current_iteration (int) – Integer value representing iteration
event (threading.Event()) – Event object to signal the execution completion

ocs_ci.helpers.odf_cephfs_snap module

ocs_ci.helpers.odf_cephfs_snap.create_provider_retain_cephfs_snapclass(snapclass_name, storage_client_name)

Create a CephFS VolumeSnapshotClass with deletionPolicy Retain.

In multicluster (provider-consumer) mode: creates the snapclass on the provider, registers it on the matching StorageConsumer so it propagates to the client, and waits for it to appear on the client cluster.

In standalone (single-cluster) mode: creates the snapclass directly on the current cluster without StorageConsumer patching.

Must be called while the active config context is the client cluster so that storage_client_name is resolved correctly before switching.

Parameters:

snapclass_name (str) – Name to give the new VolumeSnapshotClass.
storage_client_name (str) – Name of the StorageClient resource on the client cluster (used to identify the matching StorageConsumer on the provider). Unused in standalone mode.

Returns:

A no-argument teardown function that removes the snapclass.: Register it with request.addfinalizer.

Return type:

callable

ocs_ci.helpers.odf_cephfs_snap.delete_volumesnaps_volumesnapcontents(snap_list_names)

Delete every VolumeSnapshot and its VolumeSnapshotContent for each entry in snap_list_names.

The Ceph-side snapshot is retained due to the Retain deletion policy, becoming orphaned.

Parameters:: snap_list_names (list[dict]) – Snapshot data as produced by TestCephFSOrphanedSnapshotAlert.create_retain_cephfs_snapshots. Each dict must contain a "snap_obj" key holding the k8s VolumeSnapshot object.

ocs_ci.helpers.odf_cephfs_snap.get_cephfs_snap_by_name(snap_entries, cephfs_snap_name)

Return the snap entry whose Ceph-side snapshot name matches.

Parameters:

snap_entries (list[dict]) – Output of get_cephfs_snap_entries().
cephfs_snap_name (str) – Ceph snapshot name to search for (the snapshot column, e.g. csi-snap-<uuid>).

Returns:

Matching entry.

Return type:

dict

Raises:

AssertionError – If no entry with the given name is found.

ocs_ci.helpers.odf_cephfs_snap.get_cephfs_snap_entries(snap_runner)

Run odf cephfs-snap ls and return the parsed snapshot entries.

Parameters:

snap_runner – Runner object with an ls() method that returns a subprocess result (stdout as bytes).

Returns:

Parsed entries; empty list if no snapshots exist.: Each dict has keys: filesystem, subvolume, subvolumegroup, snapshot, state.

Return type:

list[dict]

ocs_ci.helpers.odf_cephfs_snap.parse_snap_ls(output)

Parse the tabular output of odf cephfs-snap ls into a list of dicts.

Parameters:

output (str) – stdout from odf cephfs-snap ls.

Returns:

Each dict has keys: filesystem, subvolume,: subvolumegroup, snapshot, state.

Return type:

list[dict]

ocs_ci.helpers.odf_cli module

class ocs_ci.helpers.odf_cli.ODFCLICephfsSnapRunner(storage_client=None, rados_namespace=None, svg=None, filesystem=None)

Bases: ODFCliRunner

Initializes and returns an instance of ODFCliRunner. ODFCliRunner subclass for odf cephfs-snap subcommands.

Holds instance-level defaults for the global optional flags shared by all cephfs-snap subcommands (–storage-client, –rados-namespace, –svg, –filesystem). None means “omit the flag and let the CLI use its own default”. The namespace flag is intentionally excluded here because ODFCliRunner.run_command already prepends -n {cluster_namespace}.

Parameters:

storage_client (str) – StorageClient CR name. Defaults to config.ENV_DATA[“storage_client_name”] when None.
rados_namespace (str) – Rados namespace for omap operations.
svg (str) – Subvolume group name.
filesystem (str) – CephFS filesystem name.

delete(subvolume, snapshot, storage_client=None, rados_namespace=None, svg=None, filesystem=None)

Run odf cephfs-snap delete <subvolume> <snapshot>.

Parameters:

subvolume (str) – Subvolume name.
snapshot (str) – Snapshot name.

Returns:

result from exec_cmd.

Return type:

CompletedProcess

ls(storage_client=None, rados_namespace=None, svg=None, filesystem=None, orphaned=False)

Run odf cephfs-snap ls and return the raw command result.

Parameters:: orphaned (bool) – If True, pass –orphaned to list only orphaned snapshots.
Returns:: result from exec_cmd.
Return type:: CompletedProcess

run_command(command, storage_client=None, rados_namespace=None, svg=None, filesystem=None, orphaned=False)

class ocs_ci.helpers.odf_cli.ODFCLIRetriever

Bases: object

add_cli_to_path(): Add the directory containing the ODF CLI binary to the system PATH.

check_odf_cli_binary()

Check if the ODF CLI binary exists, is executable, and matches the cluster version.

Returns:: True if the binary exists, is executable, and matches the cluster version, False otherwise.
Return type:: bool

retrieve_odf_cli_binary()

Download and set up the ODF-CLI binary.

Raises:: NotSupportedException – If ODF CLI is not supported on the current version or deployment.

class ocs_ci.helpers.odf_cli.ODFCliRunner

Bases: object

get_recovery_profile()

Retrieve the current recovery profile using the ODF CLI.

Returns:: The name of the current recovery profile (e.g., ‘low_recovery_ops’, ‘balanced’, ‘high_recovery_ops’).
Return type:: str

Notes

If the CLI returns no output, the method logs a warning and returns the default profile ‘balanced’ as a fallback.

run_command(command_args: str | list) → str

run_get_health()

run_get_mon_endpoint()

run_get_recovery_profile()

run_help()

run_maintenance_start(deployment_name)

This starts the maintenance mode for the deployment.

Parameters:

deployment_name (str) – Name of the deployment that you want
i.e (it to be in maintenance mode) –
deployments (either Mon or OSD) –

Raises:

CommandFailed – If the CLI command fails.

run_maintenance_stop(deployment_name)

This stops the maintenance mode for the deployment.

Parameters:

deployment_name (str) – Name of the maintenance mode deployment
stopped. (that you want it to be) –

Raises:

CommandFailed – If the CLI command fails.

run_noobaa(command_args: str | list, namespace: str = None, use_yes: bool = False, ignore_error: bool = False, **kwargs) → str

Run noobaa subcommand via odf-cli.

Parameters:

command_args – NooBaa command arguments (without ‘noobaa’ prefix) Can be string or list
namespace – Override default namespace (if provided)
use_yes – If True, pipe ‘yes’ to the command for auto-confirmation
ignore_error – If True, don’t raise exception on non-zero exit
**kwargs – Additional arguments to pass to exec_cmd

Returns:

Command output from exec_cmd

Examples

run_noobaa(“status”) run_noobaa(“obc list”) run_noobaa([“bucket”, “list”], namespace=”my-namespace”)

run_object_disable_remote_obc()

Disable remote OBC on client cluster.

This command is used in Provider/Client (HCI) deployments to disable Object Bucket Claims (OBC) on the client cluster.

Raises:: CommandFailed – If the CLI command fails.

run_object_enable_remote_obc()

Enable remote OBC on client cluster.

This command is used in Provider/Client (HCI) deployments to enable Object Bucket Claims (OBC) on the client cluster.

Raises:: CommandFailed – If the CLI command fails.

run_rook_restart()

run_rook_set_log_level(log_level: str)

run_set_ceph_log_level(service: str, log_level: str, subsystem: str)

run_set_recovery_profile(profile_name)

Set the recovery profile using the ODF CLI.

Parameters:: profile_name (str) – The name of the recovery profile to apply (e.g., ‘low_recovery_ops’, ‘balanced’, ‘high_recovery_ops’).
Raises:: CommandFailed – If the CLI command fails.

run_set_recovery_profile_balanced(): Set the recovery profile to ‘balanced’.

run_set_recovery_profile_high(): Set the recovery profile to ‘high_recovery_ops’.

run_set_recovery_profile_low(): Set the recovery profile to ‘low_recovery_ops’.

ocs_ci.helpers.odf_cli.odf_cli_cephfs_snap_setup_helper(**kwargs)

Initialize and return an ODFCLICephfsSnapRunner with the ODF CLI binary set up.

Parameters:: **kwargs – Optional constructor arguments forwarded to ODFCLICephfsSnapRunner (storage_client, rados_namespace, svg, filesystem).
Returns:: The initialized runner.
Return type:: ODFCLICephfsSnapRunner

ocs_ci.helpers.odf_cli.odf_cli_setup_helper(odf_cli_class=<class 'ocs_ci.helpers.odf_cli.ODFCliRunner'>, **kwargs)

Initializes and returns an instance of ODFCliRunner (or a subclass). Downloads the ODF CLI binary if it does not exist.

Parameters:

odf_cli_class – Runner class to instantiate (default: ODFCliRunner).
**kwargs – Extra keyword arguments forwarded to the runner constructor.

Returns:

The initialized runner.

Return type:

ODFCliRunner

Raises:

NotSupportedException – If ODF CLI is not supported on the current version or deployment.
RuntimeError – If CLI binary download or ODFCliRunner initialization fails.

ocs_ci.helpers.osd_resize module

ocs_ci.helpers.osd_resize.base_ceph_verification_steps_post_resize_osd(old_osd_pods, old_osd_pvcs, old_osd_pvs, expected_storage_size, expected_ceph_capacity=None)

Check the Ceph verification steps post resize OSD. It will perform the following steps: 1. Check the resources state post resize OSD 2. Check the resources size post resize OSD 3. Check the Ceph state post resize OSD

Parameters:

old_osd_pods (list) – The old osd pod objects before resizing the osd
old_osd_pvcs (list) – The old osd PVC objects before resizing the osd
old_osd_pvs (list) – The old osd PV objects before resizing the osd
expected_storage_size (str) – The expected storage size after resizing the osd
expected_ceph_capacity (int) – Expected Ceph raw capacity in GiB after OSD resize

Raises:

StorageSizeNotReflectedException – If the current storage size, PVCs, PVs, and ceph capacity are not in the expected size

ocs_ci.helpers.osd_resize.basic_resize_osd(old_storage_size)

The function perform the basic resize osd scenario. It increases the osd size by multiply 2

Parameters:: old_storage_size (str) – The old storagecluster storage size(which represent the old osd size)
Returns:: The new storage size after increasing the osd size
Return type:: str

ocs_ci.helpers.osd_resize.ceph_verification_steps_post_resize_osd(old_osd_pods, old_osd_pvcs, old_osd_pvs, expected_storage_size, expected_ceph_capacity=None, num_of_tries=6)

Try to execute the function ‘base_ceph_verification_steps_post_resize_osd’ a number of tries until success, ignoring the exception ‘StorageSizeNotReflectedException’. In every iteration, if we get the exception ‘StorageSizeNotReflectedException’, it will restart the osd pods and try again until it reaches the maximum tries.

Parameters:

old_osd_pods (list) – The old osd pod objects before resizing the osd
old_osd_pvcs (list) – The old osd PVC objects before resizing the osd
old_osd_pvs (list) – The old osd PV objects before resizing the osd
expected_storage_size (str) – The expected storage size after resizing the osd
expected_ceph_capacity (int) – Expected Ceph raw capacity in GiB after OSD resize
num_of_tries (int) – The number of tries to try executing the function ‘base_ceph_verification_steps_post_resize_osd’.

Raises:

StorageSizeNotReflectedException – If the current storage size, PVCs, PVs, and ceph capacity are not in the expected size

ocs_ci.helpers.osd_resize.check_ceph_capacity_increased(expected_ceph_capacity, tolerance_percent=1.0)

Validates that the current Ceph raw capacity is within the expected range.

Parameters:

expected_ceph_capacity (int) – Expected Ceph raw capacity in GiB
tolerance_percent (float) – Acceptable deviation in percentage (default: 1%)

Raises:

StorageSizeNotReflectedException – If current capacity is outside the tolerated range

ocs_ci.helpers.osd_resize.check_ceph_health_after_resize_osd(ceph_health_tries=40, ceph_rebalance_timeout=900)

Check Ceph health after resize osd

Parameters:

ceph_health_tries (int) – The number of tries to wait for the Ceph health to be OK.
ceph_rebalance_timeout (int) – The time to wait for the Ceph cluster rebalanced.

ocs_ci.helpers.osd_resize.check_ceph_state_post_resize_osd()

Check the Ceph state post resize osd. The function checks the Ceph device classes and osd tree.

Raises:: CephHealthException – In case the Ceph device classes and osd tree checks didn’t finish successfully

ocs_ci.helpers.osd_resize.check_resize_osd_pre_conditions(expected_storage_size)

Check the resize osd pre-conditions: 1. Check that the current storage size is less than the osd max size 2. If we use AWS or ROSA HCP platforms, check that the osd resize count is no more than the AWS max resize count.

If the conditions are not met, the test will be skipped.

Parameters:: expected_storage_size (str) – The expected storage size for the storage cluster

ocs_ci.helpers.osd_resize.check_resources_state_post_resize_osd(old_osd_pods, old_osd_pvcs, old_osd_pvs)

Check that the pods, PVCs, and PVs are in the expected state post resizing the osd. It will perform the following steps: 1. Check that the old osd pods are in a terminating state or deleted 2. Check that the new osd pods running, and we have exactly the same number of osd pods as the old ones. 3. Check that the PVCs are in a Bound state 4. Check that the old PVC and PV names are equal to the current PVC and PV names

Parameters:

old_osd_pods (list) – The old osd pod objects before resizing the osd
old_osd_pvcs (list) – The old osd PVC objects before resizing the osd
old_osd_pvs (list) – The old osd PV objects before resizing the osd

Raises:

StorageSizeNotReflectedException – If the OSD pods failed to restart
ResourceWrongStatusException – The old PVC and PV names are not equal to the current PVC and PV names

ocs_ci.helpers.osd_resize.check_storage_size_is_reflected(expected_storage_size, expected_ceph_capacity=None)

Check that the expected storage size is reflected in the current storage size, PVCs, PVs, and ceph capacity.

Parameters:

expected_storage_size (str) – The expected storage size
expected_ceph_capacity (int) – Expected Ceph raw capacity in GiB

Raises:

StorageSizeNotReflectedException – If the current storage size, PVCs, PVs, and ceph capacity are not in the expected size

ocs_ci.helpers.osd_resize.check_storage_size_is_reflected_in_ui(): Check that the current total storage size is reflected in the UI ‘ocs-storagecluster-storagesystem’ page.

ocs_ci.helpers.osd_resize.update_resize_osd_count(old_storage_size)

Update the resize osd count

Parameters:: old_storage_size (str) – The old storage size before the osd resizing

ocs_ci.helpers.performance_lib module

ocs_ci.helpers.performance_lib.calculate_operation_time(name, times)

Calculation the total time in seconds.

Parameters:

name (str) – The name of object to calculate the time - for logging only
times (dict) – Dictioanry of {‘start’: datetime, ‘end’: datetime, ‘total’: int}

Returns:

the number of seconds between start time to end time.

Return type:

float

ocs_ci.helpers.performance_lib.csi_bulk_pvc_time_measure(interface, pvc_objs, operation, start_time)

Measure PVC time (create / delete) in the CSI driver

Parameters:

interface (str) – an interface (RBD or CephFS) to run on
pvc_objs (list) – list of the PVC objects which we want to mesure
operation (str) – which operation to mesure - ‘create’ / ‘delete’
start_time (str) – Formatted time from which and on to search the relevant logs

Returns:

time in seconds which took the CSI to hendale the PVC

Return type:

(float)

ocs_ci.helpers.performance_lib.csi_pvc_time_measure(interface, pvc_obj, operation, start_time)

Measure PVC time (create / delete) in the CSI driver

Parameters:

interface (str) – an interface (RBD or CephFS) to run on
pvc_obj (PVC) – the PVC object which we want to mesure
operation (str) – which operation to mesure - ‘create’ / ‘delete’
start_time (str) – Formatted time from which and on to search the relevant logs

Returns:

time in seconds which took the CSI to hendale the PVC

Return type:

(float)

ocs_ci.helpers.performance_lib.extruct_timestamp_from_log(line)

Excructing from the log line the timestamp of a message. adidng the current year since it is not exists in the log line.

Parameters:: line (str) – a log line.
Returns:: string of the timestamp from the log line.
Return type:: str

ocs_ci.helpers.performance_lib.get_logfile_names(interface, provisioning=True)

Finds names for log files pods in which logs for pvc creation are located

Parameters:

interface (str) – an interface (RBD or CephFS) to run on
provisioning (bool) – if True, look for the provisioner log pods

Returns:

names of the log files relevant for searching in

Return type:

log names (list)

ocs_ci.helpers.performance_lib.get_pvc_provision_times(interface, pvc_name, start_time, time_type='all', op='all')

Get the starting/ending creation time of a PVC based on provisioner logs

Parameters:

interface (str) – The interface backed the PVC
pvc_name (str / list) – Name of the PVC(s) for creation time the list will be list of pvc objects
start_time (time) – the starttime of the test to reduce log size reading
time_type (str) – the type of time to mesure : csi / total / all (csi & total)
op (str) – the operation to mesure : create / delete / all (create & delete)

Returns:

all creation and deletion times for each pvc.

Return type:

dictioanry

ocs_ci.helpers.performance_lib.get_snapshot_time(snap_name, status, start_time)

Get the starting/ending creation time of a snapshot based on logs

The time and date extraction code below has been modified to read the month and day data in the logs. This fixes an error where negative time values are calculated when test runs cross midnight. Also, previous calculations would not set the year, and so the calculations were done as if the year were 1900. This is not a problem except that 1900 was not a leap year and so the next February 29th would throw ValueErrors for the whole day. To avoid this problem, changes were made to also include the current year.

Incorrect times will still be given for tests that cross over from December 31 to January 1.

Parameters:

pvc_name (str / list) – Name of the PVC(s) for creation time the list will be list of pvc objects
status (str) – the status that we want to get - Start / End
start_time (str) – start time, starting from which the logs are searched

Returns:

Time of searched snapshot operation

Return type:

datetime object

ocs_ci.helpers.performance_lib.measure_csi_snapshot_creation_time(interface, snapshot_id, start_time)

Measure PVC creation time, provided pvc name and time after which the PVC was created

Parameters:

interface (str) – an interface (RBD or CephFS) to run on
snapshot_id (str) – Id of the snapshot which creation time is measured
start_time (str) – Formatted time from which and on to search the relevant logs

Returns:

(float) snapshot creation time in seconds

ocs_ci.helpers.performance_lib.measure_pvc_creation_time(interface, pvc_name, start_time)

Measure PVC creation time, provided pvc name and time after which the PVC was created

Parameters:

interface (str) – an interface (RBD or CephFS) to run on
pvc_name (str) – Name of the pvc for which we measure the time
start_time (str) – Formatted time from which and on to search the relevant logs

Returns:

(float) creation time for PVC in seconds

ocs_ci.helpers.performance_lib.measure_total_snapshot_creation_time(snap_name, start_time)

Measure Snapshot creation time based on logs

Parameters:

snap_name (str) – Name of the snapshot for creation time measurement
start_time (str) – start time, starting from which the logs are searched

Returns:

Creation time for the snapshot

Return type:

float

ocs_ci.helpers.performance_lib.pod_attach_csi_time(interface, pv_name, start_time, namespace='openshift-storage')

Get the pod start/attach csi time of a pod based on csi-rbdplugin container in csi-rbdplugin pods

Parameters:

interface (str) – The interface backed the PVC
pv_name (str) – Name of the PV
start_time (time) – the start time of the test to reduce log size reading
namespace (str) – the tests namespace

Returns:

Pod attachment csi time in seconds (time. time): Start time of node stage, End time of node publish

Return type:

float

Raises:

Exception – in case that the expected time logs are not found

ocs_ci.helpers.performance_lib.pod_bulk_attach_csi_time(interface, pvc_objs, csi_start_time, namespace)

Parameters:

interface (str) – The interface backed the PVC
pvc_objs (list) – List of PVC objects to which pods were attached
csi_start_time (time) – the start time of the test to reduce log size reading
namespace (str) – the tests namespace

Returns:

ocs_ci.helpers.performance_lib.read_csi_logs(log_names, container_name, start_time)

Reading specific CSI logs starting on a specific time

Parameters:

log_names (list) – list of pods to read log from them
container_name (str) – the name of the specific container in the pod
start_time (time) – the time stamp which will use as starting point in the log

Returns:

list of lines from all logs

Return type:

list

ocs_ci.helpers.performance_lib.run_command(cmd, timeout=600, out_format='string', **kwargs)

Running command on the OS and return the STDOUT & STDERR outputs in case of argument is not string or list, return error message

Parameters:

cmd (str/list) – the command to execute
timeout (int) – the command timeout in seconds, default is 10 Min.
out_format (str) – in which format to return the output: string / list
kwargs (dict) – dictionary of argument as subprocess get

Returns:

all STDOUT and STDERR output as list of lines, or one string separated by NewLine

Return type:

list or str

ocs_ci.helpers.performance_lib.run_oc_command(cmd, namespace=None)

Running an ‘oc’ command This function is needed in Performance tests in order to be able to run a separate command within the test without creating additional objects which increases memory consumed by the test.

Parameters:

cmd (str) – the command to run
namespace (str) – the namespace where to run the command. If None is provided then value from config will be used.

Returns:

the results of the command as list of lines

Return type:

list

ocs_ci.helpers.performance_lib.string_to_time(time_string)

Converting string which present a time stamp to a time object

Parameters:: time_string (str) – the string to convert
Returns:: a time object
Return type:: datetime

ocs_ci.helpers.performance_lib.wait_for_cronjobs(namespace, cronjobs_num, msg, timeout=60)

Runs ‘oc get reclaimspacecronjob’ with the TimeoutSampler

Parameters:

namespace (str) – namespace in which cronjobs will be looked for
cronjobs_num (int) – the exact number of cronjobs that should exist
msg (str) – Error message to be printed if the desired condition is not reached
timeout (int) – Timeout

Returns:

Result of ‘oc get reclaimspacecronjob’ command

Return type:

list

ocs_ci.helpers.performance_lib.wait_for_resource_bulk_status(resource, resource_count, namespace, status, timeout=60, sleep_time=3)

Waiting for bulk of resources (from the same type) to reach the desire status

Parameters:

resource (str) – the resoure type to wait for
resource_count (int) – the number of rusource to wait for - to wait for deleteion of resources, this should be ‘0’
namespace (str) – the namespace where the resources should be
status (str) – the status of the resources to be in.
timeout (int) – how much time to wait for the resources (in sec.)- default is 1 Minute
sleep_time (int) – how much time to wait between each iteration check - default is 3 sec.

Returns:

‘True’ if all resources reach the desire state

Return type:

bool

Raises:

Exception – in case of not all resources reach the desire state.

ocs_ci.helpers.performance_lib.write_fio_on_pod(pod_obj, file_size)

Writes IO of file_size size to a pod

Parameters:

pod_obj – pod object to write IO
file_size – the size of the IO to be written opn pod

ocs_ci.helpers.pod_helpers module

ocs_ci.helpers.pod_helpers.get_all_pods_container_resource_details(pod_objs)

Extracts resource requests and limits from the pod objects.

Parameters:

pod_objs (list) – A list of pod objects from a live cluster.

Returns:

A dictionary where keys are pod prefix names and values are a list of dictionaries with: container resource details. This structure is used to group containers by their parent pod prefix.

Return type:

dict

ocs_ci.helpers.pod_helpers.get_pod_container_resource_details(pod_obj)

Extracts resource requests and limits for all containers within a single pod.

Parameters:

pod_obj (Pod) – A Pod object containing metadata and spec information.

Returns:

A list of dictionaries, where each dictionary represents a container and its resource: requests/limits. Returns an empty list if no containers are found.

Return type:

list

ocs_ci.helpers.pod_helpers.run_io_on_pods(pods, pod_file_name, size='1G', runtime=30)

Helper function to run IO on the pods

Parameters:

pods (list) – The list of pods for running the IO
pod_file_name (str) – The pod file name for fio
size (str) – Size in MB or Gi, e.g. ‘200M’. Default value is ‘1G’
runtime (int) – The number of seconds IO should run for

ocs_ci.helpers.pod_helpers.run_io_on_small_groups_of_pods(pods, pod_file_name, size='1G', runtime=30, num_of_groups=3, do_md5sum=False, wait_between_groups=30)

Run IO on pods in smaller groups to avoid overwhelming the Ceph cluster with simultaneous writes.

After each group starts, the function waits ‘wait_between_groups’ seconds before starting the next group. If do_md5sum is True, it also waits for IO to complete and calculates md5sum per group.

Parameters:

pods (list) – The list of pods for running the IO
pod_file_name (str) – The pod file name for fio
size (str) – Size in MB or Gi, e.g. ‘200M’. Default value is ‘1G’
runtime (int) – The number of seconds IO should run for
num_of_groups (int) – The number of groups to divide the pods into for running IO
do_md5sum (bool) – If True, wait for IO to complete and calculate md5sum per group
wait_between_groups (int) – Seconds to wait between starting each group

ocs_ci.helpers.pod_helpers.validate_all_pods_container_resources(pods_resources_details_dict)

Validate all live pods to ensure their container resource values exist and start with a digit.

Parameters:

pods_resources_details_dict (dict) – Mapping of pod names to container resource details.

Returns:

Dictionary with validation results.

Examples:

{
    "result": bool,  # True if all pods have valid values
    "invalid_values": dict  # Pods with invalid or missing resource values
}

Return type:

dict

ocs_ci.helpers.pod_helpers.validate_pod_container_resources(pod_name, pod_resources_details)

Validate that all container resource values in a pod exist and start with a digit.

Parameters:

pod_name (str) – Name of the pod.
pod_resources_details (list) – List of container resource details.

Returns:

Dictionary with validation results.

Examples:

{
    "result": bool,              # True if all values are valid
    "invalid_values": dict       # Containers and fields with invalid values
}

Return type:

dict

ocs_ci.helpers.proxy module

ocs_ci.helpers.proxy.get_cluster_proxies()

Get http and https proxy configuration:

If both DEPLOYMENT["proxy"] and DEPLOYMENT["disconnected"] are not set or set to false, return empty strings for all http_proxy, https_proxy and no_proxy.
Next statements apply, if one of DEPLOYMENT["proxy"] or DEPLOYMENT["disconnected"] is set to true.
If configuration ENV_DATA["http_proxy"] and ENV_DATA["no_proxy"] (and prospectively ENV_DATA["https_proxy"]) exists, return the respective values. (If https_proxy not defined, use value from http_proxy.)
If configuration ENV_DATA["http_proxy"] or ENV_DATA["no_proxy"] doesn’t exist, try to gather cluster wide proxy configuration. (If just one of those parameters exists, the configuration value have higher priority and the other value is gathered from cluster wide proxy configuration object.)
Additionally if http_proxy is gathered from cluster wide proxy configuration and DEPLOYMENT["disconnected"] is not defined or set to false, mark the cluster as cluster behind proxy by setting DEPLOYMENT["proxy"] to true.

Returns:: (http_proxy, https_proxy, no_proxy)
Return type:: tuple

ocs_ci.helpers.proxy.update_container_with_proxy_env(job_pod_dict)

If applicable, update Job or Pod configuration dict with http_proxy, https_proxy and no_proxy env variables (required for disconnected clusters and clusters behind proxy).

Parameters:: job_pod_dict (dict) – dictionary with Job or Pod configuration (updated in-place)

ocs_ci.helpers.pvc_ops module

ocs_ci.helpers.pvc_ops.create_pvcs(multi_pvc_factory, interface, project=None, status='', storageclass=None)

ocs_ci.helpers.pvc_ops.delete_pods(pod_objs): Delete pods

ocs_ci.helpers.pvc_ops.test_create_delete_pvcs(multi_pvc_factory, pod_factory, project=None)

ocs_ci.helpers.sanity_helpers module

class ocs_ci.helpers.sanity_helpers.Sanity

Bases: object

Class for cluster health and functional validations

create_pvc_delete(multi_pvc_factory, project=None): Creates and deletes all types of PVCs

create_resources(pvc_factory, pod_factory, bucket_factory, rgw_bucket_factory, run_io=True, bucket_creation_timeout=180)

Sanity validation: Create resources - pods, OBCs (RGW and MCG), PVCs (FS and RBD) and run IO

Parameters:

pvc_factory (function) – A call to pvc_factory function
pod_factory (function) – A call to pod_factory function
bucket_factory (function) – A call to bucket_factory function
rgw_bucket_factory (function) – A call to rgw_bucket_factory function
run_io (bool) – True for run IO, False otherwise
bucket_creation_timeout (int) – Time to wait for the bucket object creation.

delete_resources(): Sanity validation - Delete resources (pods, PVCs and OBCs)

health_check(cluster_check=True, tries=20, fix_ceph_health=False, update_jira=True, no_exception_if_jira_issue_updated=False)

Perform Ceph and cluster health checks

Parameters:

cluster_check (bool) – If true, perform the cluster check. False, otherwise.
tries (int) – The number of tries to perform ceph health check
fix_ceph_health (bool) – If True, it will try to fix the health to be OK even if it will recover, we will get an exception CephHealthRecoveredException
update_jira (bool) – If True, it will update the Jira issue with comment and MG logs
no_exception_if_jira_issue_updated (bool) – If True, it will not raise an exception if the Jira issue is updated and ceph health is recovered. Applicable only if fix_ceph_health is True.

obc_put_obj_create_delete(mcg_obj, bucket_factory, timeout=300): Creates bucket then writes, reads and deletes objects

class ocs_ci.helpers.sanity_helpers.SanityExternalCluster

Bases: Sanity

Helpers for health check and functional validation in External mode

class ocs_ci.helpers.sanity_helpers.SanityManagedService(create_scale_pods_and_pvcs_using_kube_job_on_ms_consumers, scale_count=None, pvc_per_pod_count=5, start_io=True, io_runtime=None, pvc_size=None, max_pvc_size=30, consumer_indexes=None)

Bases: Sanity

Class for cluster health and functional validations for the Managed Service

base_create_resources_on_ms_consumers()

Create resources on MS consumers. This function uses the factory “create_scale_pods_and_pvcs_using_kube_job_on_ms_consumers” for creating the resources - Create scale pods, PVCs, and run IO using a Kube job on MS consumers. If it fails to create the resources, it cleans up the current resources.

Raises:: ocs_ci.ocs.exceptions.CommandFailed – In case of a command failed

create_resources_on_ms_consumers(tries=1, delay=30)

Try creates resources for MS consumers ‘tries’ times with delay ‘delay’ between the iterations using the method ‘base_create_resources_on_ms_consumers’. If not specified, the default value of ‘tries’ is 1 - which means that by default, it only tries to create the resources once. In every iteration, if it fails to generate the resources, it cleans up the current resources before continuing to the next iteration.

Parameters:

tries (int) – The number of tries to create the resources on MS consumers
delay (int) – The delay in seconds between retries

Raises:

ocs_ci.ocs.exceptions.CommandFailed – In case of a command failed

delete_resources_on_ms_consumers(): Delete the resources from the MS consumers

health_check_ms(cluster_check=True, tries=20, consumers_ceph_health_check=True, consumers_tries=10)

Perform Ceph and cluster health checks on Managed Service cluster

Parameters:

cluster_check (bool) – If true, perform the cluster check. False, otherwise.
tries (int) – The number of tries to perform ceph health check
consumers_ceph_health_check (bool) – If true and the cluster is an MS provider cluster, perform ceph health check on the ms consumer clusters.
consumers_tries – The number of tries to perform ceph health check on the MS consumer clusters

class ocs_ci.helpers.sanity_helpers.SanityProviderMode(create_scale_pods_and_pvcs_using_kube_job_on_hci_clients, scale_count=None, pvc_per_pod_count=5, start_io=True, io_runtime=None, pvc_size=None, max_pvc_size=30, client_indices=None)

Bases: Sanity

Class for cluster health and functional validations for the Provider mode platform

base_create_resources_on_clients()

Create resources on clients. This function uses the factory “create_scale_pods_and_pvcs_using_kube_job_on_hci_clients” for creating the resources - Create scale pods, PVCs, and run IO using a Kube job on HCI clients. If it fails to create the resources, it cleans up the current resources.

Raises:: ocs_ci.ocs.exceptions.CommandFailed – In case of a command failed

create_resources_on_clients(tries=1, delay=30)

Try creates resources for client ‘tries’ times with delay ‘delay’ between the iterations using the method ‘base_create_resources_on_hci_clients’. If not specified, the default value of ‘tries’ is 1 - which means that by default, it only tries to create the resources once. In every iteration, if it fails to generate the resources, it cleans up the current resources before continuing to the next iteration.

Parameters:

tries (int) – The number of tries to create the resources on the clients
delay (int) – The delay in seconds between retries

Raises:

ocs_ci.ocs.exceptions.CommandFailed – In case of a command failed

delete_resources_on_clients(**kwargs)

health_check_provider_mode(cluster_check=True, tries=20, run_client_clusters_health_check=True)

Perform Ceph and cluster health checks on the cluster

Parameters:

cluster_check (bool) – If true, perform the cluster check. False, otherwise.
tries (int) – The number of tries to perform ceph health check
run_client_clusters_health_check (bool) – If true and the cluster is a provider cluster, run the cluster health check on the client clusters.

ocs_ci.helpers.storage_auto_scaler module

ocs_ci.helpers.storage_auto_scaler.delete_all_storage_autoscalers(namespace=None, wait=True, timeout=120, force=False)

Delete all StorageAutoScaler custom resources in the cluster namespace.

Parameters:

namespace (str) – The namespace of the auto-scaler resources.
wait (bool) – Whether to wait for deletion to complete.
timeout (int) – Time in seconds to wait for deletion of each resource.
force (bool) – Force deletion if standard deletion fails.

ocs_ci.helpers.storage_auto_scaler.generate_default_scaling_threshold(default_threshold=None, min_diff=7)

Generate a safe scaling threshold based on current Ceph usage.

This function calculates a default scaling threshold that avoids triggering scaling too soon. It compares: - Ceph’s overall used capacity percentage - The highest OSD’s individual usage percentage

It selects the larger of these two and ensures the scaling threshold is at least ‘min_diff’ percent higher than that usage value. If the provided default threshold is too close to the current usage, it is increased accordingly.

Parameters:

default_threshold (int | None) – Initial threshold (default: computed dynamically).
min_diff (int) – Minimum gap (in percentage points) between current usage and scaling threshold to avoid premature scaling (default: 7).

Returns:

A safe and adjusted scaling threshold percentage.

Return type:

int

ocs_ci.helpers.storage_auto_scaler.generate_fixed_scaling_threshold(fix_capacity_to_fillup: int = 280, max_scaling_threshold: int = 35) → int

Calculate a fixed scaling threshold (in percent) from a target absolute capacity.

The threshold is computed as: (fix_capacity_to_fillup / current_ceph_capacity) * 100 and then capped by the provided ‘max_scaling_threshold’.

Parameters:

fix_capacity_to_fillup (int) – Target capacity amount to ‘fill’ before scaling (default: 240). Must use the same units as ‘get_ceph_capacity()’.
max_scaling_threshold (int) – Upper bound for the computed threshold percentage (default: 35).

Returns:

The scaling threshold percentage (not exceeding ‘max_scaling_threshold’).

Return type:

int

ocs_ci.helpers.storage_auto_scaler.get_all_storage_autoscaler_names(namespace=None)

Retrieve a list of all StorageAutoScaler resource names in the cluster namespace.

Parameters:: namespace (str) – The namespace of the auto-scaler resources.
Returns:: A list of StorageAutoScaler names. Empty if none exist.
Return type:: list

ocs_ci.helpers.storage_auto_scaler.safe_teardown_delete_all_autoscalers(namespace=None)

Safely delete all StorageAutoScaler resources, ensuring Prometheus has reconciled since the last cleanup.

This function performs the following: - Reads the last cluster cleanup timestamp from config.RUN[“cleanup_cluster_time”] - If set, waits for a Prometheus reconcile timeout (default: 660 seconds) from that time - Then deletes all autoscaler resources in the given namespace

This helps prevent premature scaling or leftover metrics issues caused by fast teardown/recreate cycles.

Parameters:: namespace (str) – Namespace of the autoscaler. Defaults to ENV_DATA namespace.

ocs_ci.helpers.storage_auto_scaler.verify_autoscaler_status_not_trigger(auto_scaler_name, namespace=None)

Verify that the StorageAutoscaler does not enter ‘InProgress’ and remains NotStarted.

Parameters:

auto_scaler_name (str) – Name of the StorageAutoScaler resource
namespace (str) – Namespace of the autoscaler. Defaults to ENV_DATA namespace.

ocs_ci.helpers.storage_auto_scaler.wait_for_auto_scaler_status(expected_status, namespace=None, resource_name=None, timeout=600, sleep=10)

Wait for the StorageAutoScaler resource to reach the desired status (PHASE column).

Parameters:

expected_status (str) – The expected status value in the “PHASE” column (e.g., ‘NotStarted’, ‘InProgress’, ‘Succeeded’, ‘Failed’).
namespace (str) – The namespace of the auto-scaler resource.
resource_name (str, optional) – Name of the StorageAutoScaler resource. If not provided, the function will detect the first available one in the namespace.
timeout (int) – Maximum time in seconds to wait for the desired status. Default is 600 seconds.
sleep (int) – Interval in seconds between status checks. Default is 10 seconds.

Raises:

ResourceWrongStatusException – If no StorageAutoScaler resources are found.
TimeoutExpiredError – If the expected status is not reached within the timeout duration.

ocs_ci.helpers.stretchcluster_helper module

ocs_ci.helpers.stretchcluster_helper.check_for_logwriter_workload_pods(sc_obj, nodes=None)

Check if logwriter pods are healthy state

Parameters:

sc_obj (StretchCluster) – Stretch cluster object
nodes (Fixture) – Nodes fixture identifying the platform nodes

ocs_ci.helpers.stretchcluster_helper.recover_by_zone_restart(sc_obj, nodes)

Recover the logwriter workload pods by nodes restart if any of the known error is found in pods

Parameters:

sc_obj (StretchCluster) – StretchCluster Object
nodes (Fixture) – Nodes fixture identifying the platform nodes

ocs_ci.helpers.stretchcluster_helper.recover_from_ceph_stuck(sc_obj)

To recover from the ceph stuck issue, we need to reset the connection score for each mon

Parameters:: sc_obj (StretchCluster Object) – A StretchCluster Object

ocs_ci.helpers.stretchcluster_helper.recover_workload_pods_post_recovery(sc_obj, pods_not_running=None)

There seems to be a known issue https://bugzilla.redhat.com/show_bug.cgi?id=2244353 and this will apply the workaround to resolve that issue

Parameters:

sc_obj (StretchCluster Object) – A stretch cluster object created for the test calling this function
pods_not_running (List) – A list of Pod objects that are not in Running state

ocs_ci.helpers.stretchcluster_helper.verify_data_corruption(sc_obj, logreader_workload_factory)

Verify if there is any data corruption in both logwriter cephfs and rbd data

Parameters:

sc_obj (Stretchcluster) – Stretch cluster object
logreader_workload_factory (pytest.fixture) – Logreader workload factory fixture

ocs_ci.helpers.stretchcluster_helper.verify_data_loss(sc_obj)

Verify data loss for both cephfs and rbd workloads

Parameters:: sc_obj (Stretchcluster) – Stretch cluster object

ocs_ci.helpers.stretchcluster_helper.verify_vm_workload(vm_obj, md5sum_before)

Validate vm workload data, new data creation and data copy-back

Parameters:

vm_obj (VirtualMachine) – VirtualMachine object
md5sum_before (String) – checksum value calculated
failure (before) –

ocs_ci.helpers.tlsprofile_helper module

Helper for TLSProfile custom resources (ocs.openshift.io/v1) and in-cluster TLS scanning via scan_cluster(). The scan logic lives in scripts/bash/tls_scan_endpoints.sh (loaded at runtime).

References (DF 4.22+): TLSProfile centralizes TLS version, ciphers, and groups for NooBaa and RGW; CR name ocs-tls-profile in the operator namespace; ocs-tls-profiles is an OLM dependency (include in disconnected mirroring). Cipher/group sets follow the product-supported lists (Mozilla Intermediate/Modern plus PQC groups). On FIPS-enabled clusters, PQ hybrids and ChaCha are not FIPS 140-2 approved; use the skipif_fips_enabled pytest mark on tests that rely on those algorithms.

class ocs_ci.helpers.tlsprofile_helper.TLSProfile(name='ocs-tls-profile', namespace=None)

Bases: object

Manage TLSProfile CRs in the ODF namespace.

API_VERSION = 'ocs.openshift.io/v1'

KIND = 'TLSProfile'

create_tls_profile(selectors=None, tls_version='TLSv1.3', ciphers=None, groups=None, do_reload=True)

Create a TLSProfile with one rule; selectors, TLS version, ciphers, and groups are configurable.

Parameters:

selectors (list | str) – Rule selectors; default is a single wildcard.
tls_version (str) – spec.rules[].config.version
ciphers (list) – spec.rules[].config.ciphers
groups (list) – spec.rules[].config.groups
do_reload (bool) – Reload OCS object after create.

Returns:

The created TLSProfile object.

Return type:

OCS

delete_tls_profile(wait=True, force=False)

Delete the TLSProfile from the cluster.

Parameters:

wait (bool) – Wait for deletion to complete.
force (bool) – Force delete with grace period 0.

Returns:

Parsed oc delete output when YAML; depends on OCP.delete behavior.

Return type:

dict

get_config_version(): Return spec.rules[0].config.version if present.

get_rule_config(): Return the first rule’s config dict or None.

get_tls_profile(out_yaml_format=True)

Fetch the TLSProfile from the cluster.

Parameters:: out_yaml_format (bool) – Return parsed YAML dict when True.
Returns:: Resource from oc get.
Return type:: dict or str

is_tls_profile_available(silent=True)

Return True if the TLSProfile exists on the cluster.

Parameters:: silent (bool) – If True, suppress warnings on failed get attempts.

replace_rules(selectors, tls_version, ciphers, groups)

Merge-patch the full rules list (single rule) on this TLSProfile.

Parameters:

selectors (list) – Rule selectors.
tls_version (str) – e.g. TLSv1.2, TLSv1.3
ciphers (list) – Cipher suite names for that version.
groups (list) – Group names for that version.

ocs_ci.helpers.tlsprofile_helper.assert_no_tls_errors_in_relevant_pod_logs(namespace, component, since='45m', tail=800, max_lines_per_pod=30)

Fail the test if recent operator / workload logs contain likely TLS error lines.

Parameters:

namespace (str) – Storage namespace (e.g. openshift-storage).
component (str) – Test parametrization key: all, noobaa, or rgw.
since (str) – Passed to oc logs --since (recent window for this run).
tail (str|int) – Max tail lines per pod.
max_lines_per_pod (int) – Cap lines included in failure output.

ocs_ci.helpers.tlsprofile_helper.assert_tls_scan_results_include_version(results, api_tls_version, *, min_matching_endpoints=1, context='')

Fail unless at least min_matching_endpoints scan rows with status == "OK" list the TLS version matching api_tls_version (see scan_cluster()).

Other OK rows may negotiate only other versions (e.g. TLS 1.2 on some ports while the profile allows 1.3); mixed ports on the same workload do not fail the check as long as the minimum number of matches is met.

Rows with NO_TLS or SKIPPED are ignored.

Parameters:

results – Return value of scan_cluster().
api_tls_version – e.g. TLSv1.2 or TLSv1.3.
min_matching_endpoints – Minimum count of OK rows that must include the expected version (default 1).
context – Short string appended to failure messages for debugging.

On success with api_tls_version TLSv1.3, each matching endpoint is logged at INFO (pod IP, port, namespace, pod, container, process, tls_versions).

ocs_ci.helpers.tlsprofile_helper.gather_tls_relevant_pod_names(namespace, component)

Pod names to scan for TLS-related log errors based on test parametrization.

Always includes ocs-operator and rook-ceph-operator; adds NooBaa / RGW pods when those paths are under test.

ocs_ci.helpers.tlsprofile_helper.get_cephobjectstore_security(namespace, name): Return CephObjectStore spec.security dict or None.

ocs_ci.helpers.tlsprofile_helper.get_first_cephobjectstore_name(namespace): Return the name of the first CephObjectStore in namespace, or None.

ocs_ci.helpers.tlsprofile_helper.get_noobaa_api_server_security(namespace): Return NooBaa spec.security.apiServerSecurity dict or None.

ocs_ci.helpers.tlsprofile_helper.scan_cluster(component='all', kubeconfig=None, namespaces=None, timeout=5, skip_ports=None, tls_versions=None, tls12_ciphers=None, tls12_groups=None, tls13_ciphers=None, tls13_groups=None, scanner_image=None, scan_timeout=600, cleanup=True)

Discover pod container ports in the storage namespace(s), run a short-lived scanner pod in scantls-system, and return per-endpoint TLS probe results.

Parameters:

component – noobaa, rgw, ceph, csi, or all.
kubeconfig – Path to kubeconfig; defaults from RUN / ENV_DATA (see _resolve_tls_scan_kubeconfig()).
namespaces – Namespaces to scan; default cluster_namespace or openshift-storage.
timeout – Per-openssl-probe timeout (seconds).
skip_ports – Comma-separated ports to skip.
tls_versions – Comma-separated versions to test (default tls1.2,tls1.3).
tls12_ciphers – Comma-separated OpenSSL cipher names for TLS 1.2.
tls12_groups – Comma-separated groups for TLS 1.2.
tls13_ciphers – Comma-separated ciphersuites for TLS 1.3.
tls13_groups – Comma-separated groups for TLS 1.3.
scanner_image – Scanner container image (default TLS_SCANNER_IMAGE).
scan_timeout – Max seconds for the remote scan.sh run.
cleanup – When True, delete the scanner pod and remove the scantls-system namespace (and all objects in it) when finished.

Returns:

One dict per endpoint with keys pod_namespace, pod_name, pod_ip, container_name, port (int), process, status (OK|NO_TLS|SKIPPED), tls_versions, tls12_ciphers, tls12_groups, tls13_ciphers, tls13_groups, reason.

Return type:

list

ocs_ci.helpers.tlsprofile_helper.scan_pod_logs_for_tls_errors(pod_name, namespace, since='30m', tail=800, unreadable_pods=None)

Return log lines that match TLS-related error heuristics for a single pod.

If unreadable_pods is None, failures from get_pod_logs() propagate. When a list is passed, the pod name and exception are appended and an empty hit list is returned so callers can fail after scanning other pods.

ocs_ci.helpers.tlsprofile_helper.tls_profile_api_version_to_scan_token(api_version): Return scanner tls_versions token (e.g. tls1.3) for a TLSProfile version.

ocs_ci.helpers.tlsprofile_helper.tlsprofile_crd_exists(): Return True if tlsprofiles.ocs.openshift.io CRD is installed.

ocs_ci.helpers.tlsprofile_helper.wait_for_cephobjectstore_security_cleared(namespace, cos_name, timeout=600, sleep=15): Wait until spec.security carries no TLSProfile-propagated cipher/group lists (empty or security omitted).

ocs_ci.helpers.tlsprofile_helper.wait_for_cephobjectstore_tls_ciphers_substring(namespace, cos_name, cipher_substr, timeout=600, sleep=15): Wait until RGW object’s spec.security.ciphers mentions cipher_substr (TLS 1.2 probe).

ocs_ci.helpers.tlsprofile_helper.wait_for_noobaa_api_server_security_absent(namespace, timeout=600, sleep=15): Wait until NooBaa has no apiServerSecurity (TLSProfile-based config cleared).

ocs_ci.helpers.tlsprofile_helper.wait_for_noobaa_tls_min_version_substring(namespace, expected_substring, timeout=600, sleep=15): Wait until NooBaa apiServerSecurity.tlsMinVersion contains expected_substring (e.g. ‘1.2’ or ‘1.3’).

ocs_ci.helpers.tlsprofile_helper.wait_for_tlsprofile_config_version(tls_profile, expected_version, timeout=600, sleep=15): Wait until TLSProfile spec shows the given TLS version string.

ocs_ci.helpers.vdbench_helpers module

ocs_ci.helpers.vdbench_helpers.auto_add_openflags_if_raw_device(config)

Automatically add ‘openflags=o_direct’ if lun path starts with ‘/dev/’.

Parameters:: config (dict) – Vdbench configuration dictionary
Returns:: Modified configuration dictionary
Return type:: dict

ocs_ci.helpers.vdbench_helpers.cleanup_vdbench_resources(namespace, deployment_name)

Clean up Vdbench resources manually.

Parameters:

namespace – Kubernetes namespace
deployment_name – Name of the deployment to clean up

ocs_ci.helpers.vdbench_helpers.create_temp_config_file(vdbench_config)

Create a temporary YAML configuration file from a dictionary.

Parameters:: vdbench_config (dict) – Vdbench configuration dictionary
Returns:: Path to temporary configuration file
Return type:: str

ocs_ci.helpers.vdbench_helpers.create_vdbench_block_scenario(vdbench_workload_factory, pvc_factory, vdbench_block_config, storage_class=None, pvc_size='10Gi', auto_start=True)

Helper function to create a block device testing scenario.

Parameters:

storage_class (str) – StorageClass name to use for the PVC. Defaults to None.
pvc_size (str) – Size of the PVC to create. Defaults to “10Gi”.
auto_start (bool) – Whether to automatically start the workload. Defaults to True.

Returns:

A tuple containing:

pvc: The created PersistentVolumeClaim object.
workload: The created Vdbench workload object.

Return type:

tuple

ocs_ci.helpers.vdbench_helpers.create_vdbench_config_from_dict(vdbench_dict)

Create a Vdbench configuration file from a dictionary.

Parameters:: vdbench_dict – Dictionary containing Vdbench configuration
Returns:: Path to created configuration file
Return type:: str

ocs_ci.helpers.vdbench_helpers.create_vdbench_performance_report(metrics_history, output_file=None)

Create a performance report from collected metrics.

Parameters:

metrics_history – Metrics history from monitoring
output_file – Path to save report file

Returns:

Performance report summary

Return type:

dict

ocs_ci.helpers.vdbench_helpers.create_vdbench_performance_scenario(vdbench_workload_factory, pvc_factory, vdbench_performance_config, storage_class=None, pvc_size='20Gi', auto_start=True)

Helper function to create a performance testing scenario.

Parameters:

storage_class (str) – StorageClass name to use for the PVC. Defaults to None.
pvc_size (str) – Size of the PVC to create. Defaults to “20Gi”.
auto_start (bool) – Whether to start the workload automatically. Defaults to True.

Returns:

A tuple containing:

pvc: The created PersistentVolumeClaim object.
workload: The created Vdbench workload object.

Return type:

tuple

ocs_ci.helpers.vdbench_helpers.create_vdbench_rwx_scenario(vdbench_workload_factory, pvc_factory, vdbench_rwx_shared_config, storage_class=None, pvc_size='15Gi', replica_count=3, auto_start=True)

Helper function to create a ReadWriteMany shared volume scenario.

Parameters:

storage_class (str) – StorageClass name for the PVC. Defaults to None.
pvc_size (str) – Size of the PVC to create. Defaults to “15Gi”.
replica_count (int) – Number of workload replicas to scale to. Defaults to 3.
auto_start (bool) – Whether to automatically start the workload. Defaults to True.

Returns:

A tuple containing:

pvc: The created PersistentVolumeClaim object.
workload: The created Vdbench workload object.

Return type:

tuple

ocs_ci.helpers.vdbench_helpers.get_default_vdbench_configs()

Get predefined Vdbench configurations for common use cases.

Returns:: Dictionary of configuration name to config dict
Return type:: dict

ocs_ci.helpers.vdbench_helpers.get_vdbench_logs(namespace, deployment_name, container='vdbench-container', search_ioerror=False)

Get logs from all Vdbench pods in a deployment.

Parameters:

namespace – Kubernetes namespace
deployment_name – Name of the deployment
container – Container name to get logs from
search_ioerror – If True, filter logs for IOError messages

Returns:

Dictionary mapping pod name to logs

Return type:

dict

ocs_ci.helpers.vdbench_helpers.get_vdbench_pods(namespace, deployment_name)

Get list of Vdbench pod names for a deployment.

Parameters:

namespace – Kubernetes namespace
deployment_name – Name of the deployment

Returns:

List of pod names

Return type:

list

ocs_ci.helpers.vdbench_helpers.monitor_vdbench_workload(workload, interval=30, duration=300)

Monitor a running Vdbench workload and collect metrics.

Parameters:

workload – VdbenchWorkload instance
interval – Monitoring interval in seconds
duration – Total monitoring duration in seconds

Returns:

List of metric snapshots

Return type:

list

ocs_ci.helpers.vdbench_helpers.parse_vdbench_output(logs)

Parse Vdbench output logs to extract performance metrics.

Parameters:: logs – Vdbench output logs
Returns:: Parsed metrics or None if parsing fails
Return type:: dict or None

ocs_ci.helpers.vdbench_helpers.validate_vdbench_config(config_dict)

Validate Vdbench configuration dictionary.

Parameters:: config_dict – Vdbench configuration to validate
Returns:: True if configuration is valid
Return type:: bool
Raises:: ValueError – If configuration is invalid

ocs_ci.helpers.vdbench_helpers.validate_vdbench_workload_health(workload, timeout=300)

Validate that a Vdbench workload is healthy and running.

Parameters:

workload – VdbenchWorkload instance to validate
timeout – Timeout for validation in seconds

Returns:

True if workload is healthy, False otherwise

Return type:

bool

ocs_ci.helpers.vdbench_helpers.wait_for_vdbench_pods_ready(namespace, deployment_name, timeout=300)

Wait for Vdbench pods to be ready.

Parameters:

namespace – Kubernetes namespace
deployment_name – Name of the deployment
timeout – Timeout in seconds

Returns:

True if pods are ready, False if timeout

Return type:

bool

ocs_ci.helpers.virtctl module

ocs_ci.helpers.virtctl.get_virtctl_tool(): Download and install virtctl tool.

ocs_ci.helpers package

Submodules

ocs_ci.helpers.bdi_helpers module

ocs_ci.helpers.ceph_helpers module

ocs_ci.helpers.ceph_mon_healthcheck module

ocs_ci.helpers.cephfs_stress_helpers module

ocs_ci.helpers.cluster_exp_helpers module

ocs_ci.helpers.cnv_helpers module

ocs_ci.helpers.disconnected module

ocs_ci.helpers.disruption_helpers module

ocs_ci.helpers.dr_helpers module

ocs_ci.helpers.dr_helpers_ui module

ocs_ci.helpers.e2e_helpers module

ocs_ci.helpers.github module

ocs_ci.helpers.helpers module

ocs_ci.helpers.keyrotation_helper module

ocs_ci.helpers.longevity_helpers module

ocs_ci.helpers.managed_services module

ocs_ci.helpers.mcg_stress_helper module

ocs_ci.helpers.odf_cephfs_snap module

ocs_ci.helpers.odf_cli module

ocs_ci.helpers.osd_resize module

ocs_ci.helpers.performance_lib module

ocs_ci.helpers.pod_helpers module

ocs_ci.helpers.proxy module

ocs_ci.helpers.pvc_ops module

ocs_ci.helpers.sanity_helpers module

ocs_ci.helpers.storage_auto_scaler module

ocs_ci.helpers.stretchcluster_helper module

ocs_ci.helpers.tlsprofile_helper module

ocs_ci.helpers.vdbench_helpers module

ocs_ci.helpers.virtctl module

Module contents