ocs_ci.utility package

Subpackages

• ocs_ci.utility.perf_dash package
  • Submodules
  • ocs_ci.utility.perf_dash.dashboard_api module
    • PerfDash
      • PerfDash.add_results()
      • PerfDash.cleanup()
      • PerfDash.connect()
      • PerfDash.get_build_id()
      • PerfDash.get_id_by_name()
      • PerfDash.get_name_by_id()
      • PerfDash.get_next_sample()
      • PerfDash.get_platform_id()
      • PerfDash.get_platform_name()
      • PerfDash.get_results()
      • PerfDash.get_test_id()
      • PerfDash.get_test_name()
      • PerfDash.get_topology_id()
      • PerfDash.get_topology_name()
      • PerfDash.get_version_builds()
      • PerfDash.get_version_id()
      • PerfDash.get_version_name()
      • PerfDash.insert_build()
      • PerfDash.insert_single_value()
  • Module contents
• ocs_ci.utility.spreadsheet package
  • Submodules
  • ocs_ci.utility.spreadsheet.spreadsheet_api module
    • GoogleSpreadSheetAPI
      • GoogleSpreadSheetAPI.get_cell_value()
      • GoogleSpreadSheetAPI.insert_row()
      • GoogleSpreadSheetAPI.print_sheet()
      • GoogleSpreadSheetAPI.update_sheet()
  • Module contents
• ocs_ci.utility.tests package
  • Submodules
  • ocs_ci.utility.tests.conftest module
    • setup_logging()
  • ocs_ci.utility.tests.test_azure_utils module
    • test_azureutil_init_noresourcegroup()
    • test_azureutil_init_resourcegroup()
    • test_azureutil_init_resourcegroup_config()
  • ocs_ci.utility.tests.test_deployment_utils module
    • test_get_cluster_prefix()
  • ocs_ci.utility.tests.test_gcp module
    • test_gcputil_init_region()
    • test_gcputil_init_region_config()
  • ocs_ci.utility.tests.test_prometheus module
    • query_range_result_bad_last_90s()
    • query_range_result_delay_60s()
    • query_range_result_ok()
    • query_range_result_single_error()
    • test_check_query_range_result_exp_delay()
    • test_check_query_range_result_exp_good_time()
    • test_check_query_range_result_exp_metric_num()
    • test_check_query_range_result_null()
    • test_check_query_range_result_simple()
    • test_check_query_range_result_simple_fail()
    • test_check_query_range_result_single_error()
  • ocs_ci.utility.tests.test_rgwutils module
    • test_get_rgw_count()
  • ocs_ci.utility.tests.test_utils module
    • A
    • B
    • test_filter_unrepresentable_values()
    • test_get_attr_chain()
    • test_get_empty_attr()
    • test_get_none_obj_attr()
    • test_get_nonexistent_attr_chain()
    • test_is_base64_block_empty_string()
    • test_is_base64_block_not_base64()
    • test_is_base64_block_rejects_regular_text()
    • test_is_base64_block_short_string()
    • test_is_base64_block_valid_base64()
    • test_is_base64_block_yaml_prefixed()
    • test_mask_secret_nomatch()
    • test_mask_secret_nosecrets()
    • test_mask_secret_null()
    • test_mask_secret_simple_positive()
    • test_run_cmd_simple_negative()
    • test_run_cmd_simple_negative_ignoreerror()
    • test_run_cmd_simple_negative_with_secrets()
    • test_run_cmd_simple_positive()
    • test_run_cmd_simple_positive_with_secrets()
    • test_truncate_large_base64_empty_output()
    • test_truncate_large_base64_multiline_block()
    • test_truncate_large_base64_no_base64()
    • test_truncate_large_base64_single_large_block()
    • test_truncate_large_base64_small_block_preserved()
    • test_truncate_large_base64_small_output()
  • ocs_ci.utility.tests.test_utils_timeout_sampler module
    • test_ti_func_values()
    • test_ts_func_exception()
    • test_ts_null()
    • test_ts_one_iteration()
    • test_ts_one_iteration_big_func_runtime()
    • test_ts_one_iteration_big_sleep_time()
    • test_ts_simple_logging()
    • test_ts_simple_usecase()
    • test_ts_wait_for_value_negative()
    • test_ts_wait_for_value_positive()
  • ocs_ci.utility.tests.test_version module
    • test_compare_from_get_semantic_version()
    • test_compare_versions()
    • test_compare_versions_wrong_expression()
    • test_get_semantic_version()
    • test_get_semantic_version_string_values()
  • Module contents
• ocs_ci.utility.workloads package
  • Submodules
  • ocs_ci.utility.workloads.fio module
    • run()
    • setup()
  • ocs_ci.utility.workloads.helpers module
    • find_distro()
  • ocs_ci.utility.workloads.jenkins module
    • run()
    • setup()
  • Module contents

Submodules

ocs_ci.utility.assisted_installer module

This module is for interaction with Assisted Installer API

class ocs_ci.utility.assisted_installer.AccountsMgmtAPI(environment='production')

Bases: OpenShiftAPI

Class for interaction with Accounts mgmt API

get_pull_secret_for_current_user()

Get pull-secret for the current user

Returns:

content of the current user’s pull-secret

Return type:

dict

class ocs_ci.utility.assisted_installer.AssistedInstallerAPI(environment='production')

Bases: OpenShiftAPI

Class for interaction with Assisted Installer API

create_cluster(data)

Create (register) new cluster in Assisted Installer console

Parameters:

data (dict) – cluster configuration based on https://api.openshift.com/?urls.primaryName=assisted-service%20service#/installer/v2RegisterCluster

create_infra_env(data)

Create (register) new Infrastructure Environment in Assisted Installer console

Parameters:

data (dict) – Infrastructure environment configuration based on https://api.openshift.com/?urls.primaryName=assisted-service%20service#/installer/RegisterInfraEnv

delete_cluster(cluster_id)

Delete cluster

Parameters:

cluster_id (str) – cluster ID

delete_infra_env(infra_env_id)

Delete Infrastructure Environment

Parameters:

infra_env_id (str) – Infra environment ID

download_cluster_file(cluster_id, dest_dir, file_name='metadata.json')

Download cluster related files

Parameters:

• cluster_id (str) – cluster ID

• dest_dir (str) – destination directory, where to place the file

• file_name (str) – file to download [metadata.json, bootstrap.ign, master.ign, worker.ign, install-config.yaml, custom_manifests.json, custom_manifests.yaml] (default: metadata.json)

download_cluster_logs(cluster_id, log_dir)

Get cluster logs

Parameters:

• cluster_id (str) – cluster ID

• log_dir (str) – destination directory, where to place the logs

Returns:

the path of the downloaded file

Return type:

str

download_infra_file(infra_env_id, dest_dir, file_name, ipxe_script_type=None, discovery_iso_type=None)

Download Infrastructure Environment related file

Parameters:

• infra_env_id (str) – Infra environment ID

• dest_dir (str) – destination directory, where to place the file

• file_name (str) – file to download [discovery.ign, ipxe-script, static-network-config]

• ipxe_script_type (str) – None or specify the script type to be served for iPXE [‘discovery-image-always’, ‘boot-order-control’]

• discovery_iso_type (str) – None or overrides the ISO type for the disovery ignition [‘full-iso’, ‘minimal-iso’]

Returns:

the path of the downloaded file

Return type:

str

get_cluster(cluster_id)

Get information about one cluster.

Parameters:

cluster_id (str) – cluster ID

Returns:

the cluster configuration

Return type:

dict

get_cluster_admin_credentials(cluster_id)

Get admin credentials and console URL.

Parameters:

cluster_id (str) – cluster ID

Returns:

the credentials and console URL

Return type:

dict

get_cluster_host(cluster_id, host_id)

Get information about host connected to cluster.

Parameters:

• cluster_id (str) – cluster ID

• host_id (str) – host ID

Returns:

the host configuration

Return type:

dict

get_cluster_hosts(cluster_id)

Get information about hosts connected to cluster.

Parameters:

cluster_id (str) – cluster ID

Returns:

the hosts configuration

Return type:

dict

get_cluster_kubeconfig(cluster_id)

Get kubeconfig

Parameters:

cluster_id (str) – cluster ID

Returns:

the kubeconfig content

Return type:

str

get_clusters()

Get list of clusters.

Returns:

the clusters configuration

Return type:

dict

get_component_versions()

Get component versions. Could be used to verify access.

Returns:

the versions

Return type:

dict

get_discovery_iso_url(infra_env_id)

Get Assisted Installer discovery iso url

Parameters:

infra_env_id (str) – Infra environment ID

get_infra_env(infra_env_id)

Get information about one infra_env.

Parameters:

infra_env_id (str) – Infra environment ID

Returns:

the infra_env configuration

Return type:

dict

get_infra_env_host(infra_env_id, host_id)

Get information about host from infra_env.

Parameters:

• infra_env_id (str) – Infra environment ID

• host_id (str) – host ID

Returns:

the host configuration

Return type:

dict

get_infra_env_hosts(infra_env_id)

Get information about hosts from infra_env.

Parameters:

infra_env_id (str) – Infra environment ID

Returns:

the hosts configuration

Return type:

dict

get_infra_envs()

Get list of infra_envs.

Returns:

the infra_envs configuration

Return type:

dict

install_cluster(cluster_id)

Launch installation of the OCP cluster.

Parameters:

cluster_id (str) – cluster ID

Returns:

the cluster configuration

Return type:

dict

update_infra_env_host(infra_env_id, host_id, update_data)

Update host configuration

Parameters:

• infra_env_id (str) – Infra environment ID

• host_id (str) – Host ID

• update_data (dict) – the data to be updated

class ocs_ci.utility.assisted_installer.OpenShiftAPI(environment='production')

Bases: object

Common class for interaction with various OpenShift APIs

property api_base_url

Prepare and return base URL for API

delete_request(endpoint, json=True, ignore_failure=False)

send DELETE request to the API endpoint

Parameters:

• endpoint (str) – the final part of the endpoint URL

• json (bool) – controls if the response should be decoded as json data

• ignore_failure (bool) – controls how to deal with failed request (if raise exception or not)

Returns:

the response from the server (if json == True, parsed as json)

Return type:

Response or dict

get_api_endpoint_url(endpoint='')

prepare and return full api endpoint URL

Parameters:

endpoint (str) – the final part of the endpoint URL

Returns:

full API endpoint URL

Return type:

str

get_request(endpoint, params=None, json=True, ignore_failure=False)

send GET request to the API endpoint

Parameters:

• endpoint (str) – the final part of the endpoint URL

• params (dict) – optional parameters encoded to the request URL

• json (bool) – controls if the response should be decoded as json data

• ignore_failure (bool) – controls how to deal with failed request (if raise exception or not)

Returns:

the response from the server (if json == True, parsed as json)

Return type:

Response or dict

property headers

prepare headers for API requests

patch_request(endpoint, data=None, json=True, ignore_failure=False)

send PATCH request to the API endpoint

Parameters:

• endpoint (str) – the final part of the endpoint URL

• data (dict) – optional data send to the request URL

• json (bool) – controls if the response should be decoded as json data

• ignore_failure (bool) – controls how to deal with failed request (if raise exception or not)

Returns:

the response from the server (if json == True, parsed as json)

Return type:

Response or dict

post_request(endpoint, data=None, json=True, ignore_failure=False)

send POST request to the API endpoint

Parameters:

• endpoint (str) – the final part of the endpoint URL

• data (dict) – optional data send to the request URL

• json (bool) – controls if the response should be decoded as json data

• ignore_failure (bool) – controls how to deal with failed request (if raise exception or not)

Returns:

the response from the server (if json == True, parsed as json)

Return type:

Response or dict

property token

Property for obtaining API token (based on OFFLINE_TOKEN)

ocs_ci.utility.aws module

class ocs_ci.utility.aws.AWS(region_name=None)

Bases: object

This is wrapper class for AWS

append_security_group(security_group_id, instance_id)

Append security group to selected ec2 nodes

Parameters:

• instance_id (str) – Instances to attach security group

• security_group_id (str) – Security group to attach

• out (print) – security group <id> added to selected nodes

attach_role_policy(role_name, policy_arn)

Attach role-policy.

Parameters:

• role_name (str) – Name of the role to attach the policy to

• policy_arn (str) – ARN of the policy to attach to the role

attach_volume(volume, instance_id, device='/dev/sdx')

Attach volume to an ec2 instance

Parameters:

• volume (Volume) – Volume instance

• instance_id (str) – id of instance where to attach the volume

• device (str) – The name of the device where to attach (default: /dev/sdx)

block_instances_access(security_group_id, instances_id)

Block ec2 instances by:

• Append security group without access permissions

• Remove original security groups

Parameters:

• security_group_id (str) – security group without access permissions

• instances_id (list) – list of ec2 instances ids

property cf_client

Property for cloudformation client

Returns:

instance of cloudformation

Return type:

boto3.client

check_volume_attributes(volume_id, name_end=None, size=None, iops=None, throughput=None, namespace=None)

Verify aws volume attributes Primarily used for faas

Parameters:

• volume_id (str) – id of the volume to be checked

• name_end (str) – expected ending of Name tag

• size (int) – expected value of volume’s size

• iops (int) – expected value of IOPS

• throughput (int) – expected value of Throughput

• namespace (str) – expected value of kubernetes.io/created-for/pvc/namespace tag

Raises:

ValueError if the actual value differs from the expected one –

cleanup_oidc_providers_by_prefix(url_prefix)

Delete all OIDC providers whose URL starts with the given prefix.

AWS has a hard limit of 100 OpenID Connect providers per account. When HCP cluster deployments fail or are aborted without proper cleanup, their OIDC providers remain and accumulate until the limit is hit, blocking all new deployments.

This method lists all OIDC providers in the account and deletes those whose issuer URL starts with the given prefix (typically the cluster’s OIDC bucket name). This safely scopes the cleanup to a single cluster’s providers without affecting other clusters.

Parameters:

url_prefix (str) – URL prefix to match, e.g. "mycluster-oidc-bucket.s3.us-west-2.amazonaws.com"

Returns:

Number of OIDC providers deleted.

Return type:

int

property cloudfront_client

Property for cloudfront client

Returns:

instance of cloudfront

Return type:

boto3.client

create_hosted_zone(cluster_name)

Create Hosted Zone

Parameters:

cluster_name (str) – Name of cluster

Returns:

Hosted Zone id

Return type:

str

create_iam_role(role_name, description, document)

Create IAM role if it doesn’t exist, or return existing role.

Parameters:

• role_name (str) – Name of the role

• description (str) – Description of the role

• document (str) – JSON string representing the role policy to assume

Returns:

Created or existing role data

Return type:

dict

create_security_group(group_name, dict_permissions, vpc_id)

Create security group with predefined group name and permissions

Parameters:

• group_name (str) – Group name (aws tag: “Group Name”)

• dict_permissions (dict) – The security group’s inbound/outbound permissions

• vpc_id (str) – For group to be attached

Returns:

newly created security group id

Return type:

str

create_stack(s3_url, index, params_list, capabilities)

Create a new cloudformation stack for worker creation

Parameters:

• s3_url (str) – An aws url for accessing s3 object

• index (int) – Integer index for stack name

• params_list (list) – of parameters (k,v) for create_stack

• capabilities (list) – of valid AWS capabilities like CAPABILITY_NAMED_IAM etc

Returns:

of (stack_name, stack_id)

Return type:

tuple

create_volume(availability_zone, name, encrypted=False, size=100, timeout=20, volume_type='gp2')

Create volume

Parameters:

• availability_zone (str) – The availability zone e.g.: us-west-1b

• name (str) – The name of the volume

• encrypted (boolean) – True if encrypted, False otherwise (default: False)

• size (int) – The size in GB (default: 100)

• timeout (int) – The timeout in seconds for volume creation (default: 20)

• volume_type (str) – ‘standard’|’io1’|’gp2’|’sc1’|’st1’ (default: gp2)

Returns:

AWS Resource instance of the newly created volume

Return type:

Volume

create_volume_and_attach(availability_zone, instance_id, name, device='/dev/sdx', encrypted=False, size=100, timeout=20, volume_type='gp2')

Create volume and attach to instance

Parameters:

• availability_zone (str) – The availability zone e.g.: us-west-1b

• instance_id (str) – The id of the instance where to attach the volume

• name (str) – The name of volume

• device (str) – The name of device where to attach (default: /dev/sdx)

• encrypted (boolean) – True if encrypted, False otherwise (default: False)

• size (int) – The size in GB (default: 100)

• timeout (int) – The timeout in seconds for volume creation (default: 20)

• volume_type (str) – ‘standard’|’io1’|’gp2’|’sc1’|’st1’ (default: gp2)

delete_all_record_sets(hosted_zone_id)

Deletes all record sets in a hosted zone

Parameters:

hosted_zone_id (str) – Hosted Zone ID example: /hostedzone/Z91022921MMOZDVPPC8D6

delete_apps_record_set(cluster_name=None, from_base_domain=False)

Delete apps record set that sometimes blocks sg stack deletion.

https://github.com/red-hat-storage/ocs-ci/issues/2549

Parameters:

• cluster_name (str) – Name of the cluster

• from_base_domain (bool) – Delete apps record set from base domain created by Flexy

delete_bucket(bucket)

Delete the bucket

Parameters:

bucket (str) – Name of the bucket to delete

delete_buckets(buckets)

Delete the buckets

Parameters:

buckets (list) – List of buckets to delete

delete_cf_stack_including_dependencies(cfs_name)

Delete cloudformation stack including dependencies.

Some of the depending resources are not deletable, so related errors are ignored and only logged. Thsi method is mainly used as a WORKAROUND for folowing Flexy issue: https://issues.redhat.com/browse/OCPQE-1521

Parameters:

cfs_name (str) – CloudFormation stack name to cleanup

delete_cloudformation_stacks(stack_names)

Delete cloudformation stacks

Parameters:

stack_names (list) – List of cloudformation stacks

delete_cloudfront_distribution(dist_id, etag)

Delete CloudFront Distribution

Parameters:

• dist_id (str) – ID of Distribution

• etag (str) – ETag header

delete_cloudfront_origin_access_identity(identity_id, etag)

Delete CloudFront Origin Access Identity

Parameters:

• identity_id (str) – ID of Origin Access Identity

• etag (str) – ETag header

delete_hosted_zone(cluster_name, delete_zone=True, delete_from_base_domain=False)

Deletes the hosted zone

Parameters:

• cluster_name (str) – Name of the cluster

• delete_zone (bool) – Whether to delete complete zone

• delete_from_base_domain (bool) – Whether to delete record from base domain

delete_iam_role(role_name)

Delete the specified IAM role.

Parameters:

role_name (str) – Name of the role to delete

delete_objects_in_bucket(bucket)

Delete objects in a bucket

Parameters:

bucket (str) – Name of the bucket to delete objects

delete_oidc_provider(provider_name)

Deletes the OIDC provider

Parameters:

provider_name (str) – OIDC provider name e.g: d3gcnqtx1lgapn.cloudfront.net

delete_record(record, hosted_zone_id)

Deletes the record from Hosted Zone

Parameters:

• record (dict) – record details to delete e.g:{ ‘Name’: ‘vavuthu-eco1.qe.rh-ocs.com.’, ‘Type’: ‘NS’, ‘TTL’: 300, ‘ResourceRecords’:[ {‘Value’: ‘ns-1389.awsdns-45.org’}, {‘Value’: ‘ns-639.awsdns-15.net’}, {‘Value’: ‘ns-1656.awsdns-15.co.uk’}, {‘Value’: ‘ns-183.awsdns-22.com’} ] }

• hosted_zone_id (str) – Hosted Zone ID example: /hostedzone/Z91022921MMOZDVPPC8D6

delete_record_from_base_domain(cluster_name, base_domain=None)

Deletes the record for cluster name in base domain

Parameters:

• cluster_name (str) – Name of the cluster

• base_domain (str) – Base domain name

delete_role_policy(role_name, policy_name)

Delete role policy.

Parameters:

• role_name (str) – Name of the role

• policy_name (str) – Name of the policy to delete

delete_s3_object(bucket_name, object_key)

Delete an object from s3 bucket

Parameters:

• bucket_name (str) – name of a valid s3 bucket

• object_key (str) – the key for s3 object

delete_security_group(security_group_id)

Delete selected security group print out: Security group <id> deleted

Parameters:

security_group_id (str) – Id of selected security group

delete_volume(volume)

Delete an ec2 volume from AWS

Parameters:

volume (Volume) – The volume to delete

detach_and_delete_volume(volume, timeout=120)

Detach volume if attached and then delete it from AWS

Parameters:

• volume (Volume) – The volume to delete

• timeout (int) – Timeout in seconds for API calls

detach_role_policy(role_name, policy_arn)

Detach role policy from IAM role.

Parameters:

• role_name (str) – Name of the role to detach the policy from

• policy_arn (str) – ARN for the policy to detach from the role

detach_volume(volume, timeout=120)

Detach volume if attached

Parameters:

• volume (Volume) – The volume to delete

• timeout (int) – Timeout in seconds for API calls

Returns:

ec2 Volume instance

Return type:

Volume

disable_cloudfront_distribution(dist_id)

Disable CloudFront Distribution

Parameters:

dist_id (str) – ID of Distribution

property ec2_client

Property for ec2 client

Returns:

instance of ec2

Return type:

boto3.client

property ec2_resource

Property for ec2 resource

Returns:

boto3.resource instance of ec2 resource

property elb_client

Property for elb client

Returns:

instance of elb client

Return type:

boto3.client

get_all_security_groups()

Get all security groups in AWS region

Returns:

All security groups

Return type:

list

get_all_stacks()

Get all the cloudformation stacks

Returns:

of all cloudformation stacks

Return type:

list

get_attached_role_policies(role_name)

Get policies attached to a role.

Parameters:

role_name (str) – Name of the role to fetch attached policies

Returns:

Attached role policies

Return type:

list

get_availability_zone_id_by_instance_id(instance_id)

Fetch availability zone out of ec2 node (EC2.Instances.placement)

Parameters:

instance_id (str) – ID of the instance - to get availability zone info from ec2 node

Returns:

availability_zone: The availability zone name

Return type:

str

get_bucket_time_based_rules(bucket_prefixes, bucket_name, hours)

Get the time bucket based prefix and hours

Parameters:

• bucket_prefixes (dict) – The rules according to them determine the number of hours the bucket can exist

• bucket_name (str) – bucket name

• hours (int) – The number of hours bucket can exist if there is no compliance with one of the rules

Returns:

The number of hours bucket can exist

Return type:

int

get_buckets_to_delete(bucket_prefix, hours)

Get the bucket with prefix which are older than given hours

Parameters:

• bucket_prefix (str) – prefix for the buckets to fetch

• hours (int) – fetch buckets that are older than to the specified number of hours

get_caller_identity()

Get STS Caller Identity Account ID

Returns:

Account ID

Return type:

str

get_caller_identity_arn()

Get STS Caller Identity ARN.

Equivalent to: aws sts get-caller-identity –query “Arn” –output text

Returns:

The ARN of the caller identity

Return type:

str

get_cloudformation_stacks(pattern)

Get cloudformation stacks

Parameters:

pattern (str) – The pattern of the stack name

get_cloudfront_distribution(dist_id)

Get CloudFront Distribusions.

Parameters:

dist_id (str) – ID of Distribution

Returns:

CloudFront Distribution

Return type:

dict

get_cloudfront_distribution_config(dist_id)

Get CloudFront Distribution config.

Parameters:

• dist_id (str) – ID of Distribution

• query (str) – Query to provide. Optional.

Returns:

Distribution config

Return type:

dict

get_cloudfront_distributions()

Get CloudFront Distribusions.

Returns:

List of CloudFront Distributions

Return type:

list

get_cloudfront_origin_access_identies()

Get CloudFront Origin Access Identities

Returns:

List of CloudFront Origin Access Identies

Return type:

list

get_cloudfront_origin_access_identity(identity_id)

Get CloudFront Origin Access Identity

Returns:

CloudFront Origin Access Identity Data

Return type:

dict

get_cluster_subnet_ids(cluster_name)

Get the cluster’s subnet ids of existing cluster

Parameters:

cluster_name (str) – Cluster name

Returns:

Subnet IDs str list

Return type:

list

get_ec2_instance(instance_id)

Get instance of ec2 Instance

Parameters:

instance_id (str) – The ID of the instance to get

Returns:

instance of ec2 instance resource

Return type:

boto3.Instance

get_ec2_instance_volumes(instance_id)

Get all volumes attached to an ec2 instance

Parameters:

instance_id (str) – The ec2 instance ID

Returns:

ec2 Volume instances

Return type:

list

get_hosted_zone_details(zone_id)

Get Hosted zone Details

Parameters:

zone_id (str) – Zone Id of cluster_name

Returns:

Response

Return type:

dict

get_hosted_zone_id(cluster_name)

Get Zone id from given cluster_name

Parameters:

cluster_name (str) – Name of cluster

Returns:

Zone id

Return type:

str

get_hosted_zone_id_for_domain(domain=None)

Get Zone id for domain

Parameters:

domain (str) – Name of the domain.

Returns:

Zone id

Return type:

str

get_iam_roles(base_name)

Retrieve a list of IAM roles that start with the base_name.

Parameters:

base_name (str) – Base of the role name

Returns:

List of IAM roles that match the base_name

Return type:

list

get_instance_id_from_private_dns_name(private_dns_name)

Get the instance id from the private dns name of the instance

Parameters:

private_dns_name (str) – The private DNS name of the instance

Returns:

The instance id associated to the private DNS name.

If not found returns None

Return type:

str

get_instance_profiles_for_role(role_name)

Get instance profiles for the specified role.

Parameters:

role_name (str) – Name of the role to find instance profiles for

Returns:

Instance Profiles for the role

Return type:

list

get_instances_by_name_pattern(pattern)

Get instances by Name tag pattern

The instance details do not contain all the values but just those we are consuming.

Those parameters we are storing for instance are: * id: id of instance * avz: Availability Zone * name: The value of Tag Name if define otherwise None * vpc_id: VPC ID * security_groups: Security groups of the instance

Parameters:

pattern (str) – Pattern of tag name like: pbalogh-testing-cluster-55jx2-worker*

Returns:

contains dictionaries with instance details mentioned above

Return type:

list

get_instances_response_by_name_pattern(pattern=None, filter_by_cluster_name=True)

Get the instances by name tag pattern. If not specified it will return all the instances, or will return the instances filtered by the cluster name.

Parameters:

• pattern (str) – Pattern of tag name like: pbalogh-testing-cluster-55jx2-worker*

• filter_by_cluster_name – Will be used only if the ‘pattern’ param not specified. If True it filters the instances by the cluster name, else if False it returns all instances.

Returns:

list of instances dictionaries.

Return type:

list

get_instances_status_by_id(instance_id)

Get instances by ID

Parameters:

instance_id (str) – ID of the instance

Returns:

The instance status

Return type:

str

get_matching_stacks(pattern)

Get only the stacks which matches the pattern

Parameters:

pattern (str) – A raw string which is re compliant

Returns:

of strings which are matching stack name

Return type:

list

get_ns_for_hosted_zone(zone_id)

Get NameServers Details from Hosted Zone

Parameters:

zone_id (str) – Zone Id of cluster_name

Returns:

NameServers

Return type:

list

get_record_sets(domain=None)

Get all the record sets in domain

Parameters:

domain (str) – Domain name to fetch the records

Returns:

list of record sets

Return type:

list

get_role_policies(role_name)

Get policies embedded in the role.

Parameters:

role_name (str) – Name of the role

Returns:

Role policies

Return type:

list

get_s3_bucket_object_url(bucket_name, object_key)

Get s3 bucket object url

Parameters:

• bucket_name (str) – Name of a valid s3 bucket

• object_key (str) – Name of the key for s3 object

Returns:

An s3 url

Return type:

s3_url (str)

get_security_groups_by_instance_id(instance_id)

Get all attached security groups of ec2 instance

Parameters:

instance_id (str) – Required instance to get security groups from it

Returns:

all_sg_ids: all attached security groups id.

Return type:

list

get_session_token(duration_seconds=7200, output_file=None)

Retrieve AWS STS session token and save it to a file.

Equivalent to:

aws sts get-session-token --duration-seconds 7200 > sts-creds.json

Parameters:

• duration_seconds (int) – Duration of the session token in seconds. Default is 7200 (2 hours). Valid range: 900 (15 min) to 129600 (36 hours).

• output_file (str) – Path to the file where credentials will be saved. If not provided, creates a temp file in the system temp directory.

Returns:

Dictionary containing:

• credentials (dict): AWS credentials with keys:

  • AccessKeyId (str): AWS access key ID

  • SecretAccessKey (str): AWS secret access key

  • SessionToken (str): AWS session token

  • Expiration (str): Expiration time in ISO format

• credentials_file (str): Path to the saved file

Return type:

dict

Raises:

ClientError – If the STS call fails

get_stack_instance_id(stack_name, logical_id)

Get the instance id associated with the cloudformation stack

Parameters:

• stack_name (str) – Name of the cloudformation stack

• logical_id (str) – LogicalResourceId of the resource ex: “Worker0”

Returns:

Id of the instance

Return type:

instance_id (str)

get_stack_name_by_instance_id(instance_id)

Get the stack name by the instance id

Parameters:

instance_id (str) – The instance id

Returns:

The stack name associated to the instance id.

If not found returns None

Return type:

str

get_stack_params(stack_name, param_name)

Get value of a particular param

Parameters:

• stack_name (str) – AWS cloudformation stack name

• param_name (str) – Stack parameter name

Returns:

Parameter value

Return type:

str

get_volume_data(volume_id)

Get volume information

Parameters:

volume_id (str) – ID of the volume

Returns:

complete volume information

Return type:

dict

get_volume_tag_value(volume_data, tag_name)

Get the value of the volume’s tag

Parameters:

• volume_data (dict) – complete volume information

• tag_name (str) – name of the tag

Returns:

value of the tag or None if there’s no such tag

Return type:

str

get_volumes_by_name_pattern(pattern)

Get volumes by pattern

Parameters:

pattern (str) – Pattern of volume name (e.g. ‘cl-vol-‘)

Returns:

Volume information like id and attachments

Return type:

list

get_volumes_by_tag_pattern(tag, pattern)

Get volumes by tag pattern

Parameters:

• tag (str) – Tag name

• pattern (str) – Pattern of tag value (e.g. ‘cl-vol-‘)

Returns:

Volume information like id and attachments

Return type:

list

get_vpc_id_by_instance_id(instance_id)

Fetch vpc id out of ec2 node (EC2.Instances.vpc_id)

Parameters:

instance_id (str) – ID of the instance - to get vpc id info from ec2 node

Returns:

vpc_id: The vpc id

Return type:

str

get_worker_ignition_location(stack_name)

Get the ignition location from given stack

Parameters:

stack_name (str) – AWS cloudformation stack name

Returns:

An AWS URL ignition location

Return type:

ignition_location (str)

get_worker_instance_profile_name(stack_name)

Get the worker instance profile name

Parameters:

stack_name (str) – AWS cloudformation stack name

Returns:

instance profile name

Return type:

worker_instance_profile_name (str)

get_worker_stacks()

Get the cloudformation stacks only for workers of this cluster

Returns:

of worker stacks

Return type:

list

property iam_client

Property for iam client

Returns:

instance of iam

Return type:

boto3.client

list_buckets()

List the buckets

Returns:

List of dictionaries which contains bucket name and creation date as keys

e.g: [ {‘Name’: ‘214qpg-oidc’, ‘CreationDate’: datetime.datetime(2023, 1, 9, 11, 27, 48, tzinfo=tzutc())}, {‘Name’: ‘214rmh4-oidc’, ‘CreationDate’: datetime.datetime(2023, 1, 9, 12, 32, 8, tzinfo=tzutc())} ]

Return type:

list

property region_name

remove_role_from_instance_profile(role_name, instance_profile_name)

Remove role from instance profile.

Parameters:

• role_name (str) – Name of the role to remove from the instance profile

• instance_profile_name (str) – Name of the instance profile to remove the role from

remove_security_group(security_group_id, instance_id)

Remove security group from selected ec2 instance (by instance id) print out: security group <id> removed from selected nodes

Parameters:

• security_group_id (str) – Security group to be removed

• instance_id (str) – Instance attached with selected security group

restart_ec2_instances(instances)

Restart ec2 instances

Parameters:

instances (dict) – A dictionary of instance IDs and names to restart

restart_ec2_instances_by_stop_and_start(instances, wait=False, force=True)

Restart EC2 instances by stop and start

Parameters:

• instances (dict) – A dictionary of instance IDs and names to stop & start

• wait (bool) – True in case wait for status is needed, False otherwise

• force (bool) – True for force instance stop, False otherwise

restore_instances_access(security_group_id_to_remove, original_security_group_dict)

Restore access to instances by removing blocking security group and append original security group.

Parameters:

• security_group_id_to_remove (str) – id of the security group

• original_security_group_dict (dict) – dict with: keys: blocked instances: ec2 instances id values: list of original security groups

property route53_client

Property for route53 client

Returns:

instance of route53 client

Return type:

boto3.client

property s3_client

Property for s3 client

Returns:

boto3.client instance of s3

property s3_resource

Property for s3 resource

Returns:

boto3.resource instance of s3

start_ec2_instances(instances, wait=False)

Starting an instance

Parameters:

• instances (dict) – A dictionary of instance IDs and names to start

• wait (bool) – True in case wait for status is needed, False otherwise

stop_ec2_instances(instances, wait=False, force=True)

Stopping an instance

Parameters:

• instances (dict) – A dictionary of instance IDs and names to stop

• wait (bool) – True in case wait for status is needed, False otherwise

• force (bool) – True for force instance stop, False otherwise

store_security_groups_for_instances(instances_id)

Stored all security groups attached to selected ec2 instances

Parameters:

instances_id (list) – ec2 instance_id

Returns:

security_group_dict: keys: blocked instances: ec2_instances ids

values: list of original security groups of each instance

Return type:

dict

property sts_client

Property for sts client

Returns:

instance of sts

Return type:

boto3.client

terminate_ec2_instances(instances, wait=True)

Terminate an instance

Parameters:

• instances (dict) – A dictionary of instance IDs and names

• wait (bool) – True in case wait for status is needed, False otherwise

update_cloudfront_distribution(dist_id, etag, dist_config)

Update CloudFront Distribution

Parameters:

• dist_id (str) – ID of Distribution

• etag (str) – ETag header

• dist_config (dict) – Distribution config data

update_hosted_zone_record(zone_id, record_name, data, type, operation_type, ttl=60, raw_data=None)

Update Route53 DNS record

Parameters:

• zone_id (str) – Zone id of DNS record

• record_name (str) – Record Name without domain eg: api.apps.ocp-baremetal-auto

• data (str) – Data to be added for DNS Record

• type (str) – DNS record type

• operation_type (str) – Operation Type (Allowed Values:- Add, Delete)

• ttl (int) – Default set to 60 sec

• raw_data (list) – Data to be added as a record

Returns:

The response from change_resource_record_sets

Return type:

dict

upload_file_to_s3_bucket(bucket_name, object_key, file_path)

Upload objects to s3 bucket

Parameters:

• bucket_name (str) – Name of a valid s3 bucket

• object_key (str) – the key for the s3 object

• file_path (str) – path for the file to be uploaded

wait_for_instances_to_stop(instances)

Wait for the instances to reach status stopped

Parameters:

instances – A dictionary of instance IDs and names

Raises:

botocore.exceptions.WaiterError – If it failed to reach the expected status stopped

wait_for_instances_to_stop_or_terminate(instances)

Wait for the instances to reach statuses stopped or terminated

Parameters:

instances – A dictionary of instance IDs and names

Raises:

botocore.exceptions.WaiterError – If it failed to reach the expected statuses stopped or terminated

wait_for_instances_to_terminate(instances)

Wait for the instances to reach status terminated

Parameters:

instances – A dictionary of instance IDs and names

Raises:

botocore.exceptions.WaiterError – If it failed to reach the expected status terminated

wait_for_record_set(response_list, max_attempts=10)

Wait for Record to be created

Parameters:

• max_attempts (int) – Max Attempt’s for Waiting

• response_list (list) – List of response

exception ocs_ci.utility.aws.AWSTimeoutException

Bases: Exception

exception ocs_ci.utility.aws.StackStatusError

Bases: Exception

ocs_ci.utility.aws.check_root_volume(volume)

Checks whether given EBS volume is root device or not

Parameters:

volume (dict) – EBS volume dictionary

Returns:

True if EBS volume is root device, False otherwise

Return type:

bool

ocs_ci.utility.aws.create_and_attach_ebs_volumes(worker_pattern, size=100, count=1, device_names=('sdx',))

Create volumes on workers

Parameters:

• worker_pattern (string) – Worker name pattern e.g.: cluster-55jx2-worker*

• size (int) – Size in GB (default: 100)

• count (int) – number of EBS volumes to attach to worker node, if it’s

• device_names (list) – list of the devices like [“sda”, “sdb”]. Length of list needs to match count!

Raises:

UnexpectedInput – In case the device_names length doesn’t match count.

ocs_ci.utility.aws.create_and_attach_sts_role()

Create IAM role to support STS deployments.

Returns:

Created role data

Return type:

dict

ocs_ci.utility.aws.create_and_attach_volume_for_all_workers(device_size=None, worker_suffix='worker', count=1, device_letters='ghijklmnopxyz')

Create volumes on workers

Parameters:

• device_size (int) – Size in GB, if not specified value from: config.ENV_DATA[“device_size”] will be used

• worker_suffix (str) – Worker name suffix (default: worker)

• count (int) – number of EBS volumes to attach to worker node

• device_letters (str) – device letters from which generate device names. e.g. for “abc” and if count=2 it will generate [“sda”, “sdb”]

ocs_ci.utility.aws.create_s3_bucket_for_hypershift_oidc(bucket_name, region, aws_credentials_path=None, namespace='local-cluster')

Create an S3 bucket for HyperShift OIDC provider with public read policy and create Kubernetes secret with bucket credentials.

This function:

1. Creates an S3 bucket in the specified region

2. Disables Block Public Access settings

3. Applies a public read policy to allow OIDC discovery

4. Creates a Kubernetes secret with AWS credentials and bucket info

Parameters:

• bucket_name (str) – Name of the S3 bucket to create

• region (str) – AWS region where the bucket should be created

• aws_credentials_path (str) – Path to AWS credentials file

• namespace (str) – Kubernetes namespace for the secret (default: “local-cluster”)

Returns:

Dictionary with bucket details:

• bucket_name (str): Name of the bucket

• region (str): AWS region of the bucket

• bucket_arn (str): ARN of the bucket

• location (str): URL location of the bucket

• secret_name (str): Name of the Kubernetes secret

• namespace (str): Kubernetes namespace of the secret

Return type:

dict

Raises:

ClientError – If bucket creation or policy application fails

ocs_ci.utility.aws.delete_cloudfront_origin_access_identities()

Delete CloudFront Origin Access Identities for the cluster.

ocs_ci.utility.aws.delete_cluster_buckets(cluster_name)

Delete s3 buckets corresponding to a particular OCS cluster

Parameters:

cluster_name (str) – name of the cluster the buckets belong to

ocs_ci.utility.aws.delete_sts_iam_roles(oidc_endpoint_url=None)

Delete IAM roles for the cluster.

Parameters:

oidc_endpoint_url (str) – Optional OIDC endpoint URL (serviceAccountIssuer) pre-fetched while the cluster API was still reachable. When provided, the cluster API is not contacted. When None, the function attempts to retrieve the URL from the live cluster; if the cluster is already unreachable the OIDC provider deletion is skipped with a warning.

ocs_ci.utility.aws.delete_subnet_tags(tag, *subnet_ids)

Delete tag from subnet(s) Default AWS account limitation is 50 tags per subnet

Parameters:

• tag (str) – Tag to delete

• subnet_ids (str) – One or more subnet IDs from which to delete the tag.

ocs_ci.utility.aws.destroy_volumes(cluster_name)

Destroy cluster volumes

Parameters:

cluster_name (str) – The name of the cluster

ocs_ci.utility.aws.get_cluster_region()

Get the AWS region of the current cluster from the Infrastructure CR.

Returns:

AWS region name (e.g. ‘us-westa-2’)

Return type:

str

Raises:

CommandFailed – If the oc command fails.

ocs_ci.utility.aws.get_cluster_vpc_cidr(region_name=None)

Get VPC CIDR block for the current OpenShift cluster.

This function retrieves the VPC CIDR block by: 1. Getting the infrastructure name from OpenShift cluster (oc get infrastructure cluster) 2. Querying AWS EC2 to find the VPC with the kubernetes.io/cluster/<infra-name> tag 3. Returning the VPC CIDR block

Parameters:

region_name (str) – AWS region name. If not provided, uses the region from config.ENV_DATA

Returns:

VPC CIDR block (e.g., “10.0.0.0/16”) or None if not found

Return type:

str

Raises:

Exception – If unable to get infrastructure name from OpenShift or AWS query fails

ocs_ci.utility.aws.get_data_volumes(deviceset_pvs)

Get the instance data volumes (which doesn’t include root FS)

Parameters:

deviceset_pvs (list) – PVC objects of the deviceset PVs

Returns:

ec2 Volume instances

Return type:

list

ocs_ci.utility.aws.get_instances_ids_and_names(instances)

Get the instances IDs and names according to nodes dictionary

Parameters:

instances (list) – Nodes dictionaries, returned by ‘oc get node -o yaml’

Returns:

The ID keys and the name values of the instances

Return type:

dict

ocs_ci.utility.aws.get_rhel_worker_instances(cluster_path)

Get list of rhel worker instance IDs

Parameters:

cluster_path (str) – The cluster path

Returns:

list of instance IDs of rhel workers

Return type:

list

ocs_ci.utility.aws.get_stack_name_from_instance_dict(instance_dict)

Get the stack name by the given instance dictionary from AWS

Parameters:

instance_dict (dict) – The instance dictionary from AWS

Returns:

The stack name of the given instance dictionary from AWS.

If not found returns None

Return type:

str

ocs_ci.utility.aws.get_unused_vpc_cidr(region_name=None, base_cidr='10.0.0.0', prefix_length=16)

Find an unused VPC CIDR block in the specified AWS region.

Parameters:

• region_name (str) – AWS region name. If not provided, uses the region from config.ENV_DATA

• base_cidr (str) – Base CIDR to start searching from. Default is “10.0.0.0”

• prefix_length (int) – CIDR prefix length. Default is 16

Returns:

An unused VPC CIDR block

Return type:

str

Raises:

• ClientError – If AWS API returns an error

• NoCredentialsError – If AWS credentials are not configured

• EndpointConnectionError – If cannot connect to AWS endpoint

• RuntimeError – If cannot find an unused CIDR after checking 254 candidates

ocs_ci.utility.aws.get_vpc_id_by_node_obj(aws_obj, instances)

This function getting vpc id by randomly selecting instances out of user aws deployment

Parameters:

• aws_obj (obj) – AWS() object

• instances (dict) – cluster ec2 instances objects

Returns:

vpc_id: The vpc id

Return type:

str

ocs_ci.utility.aws.terminate_rhel_workers(worker_list)

Terminate the RHEL worker EC2 instances

Parameters:

worker_list (list) – Instance IDs of rhel workers

Raises:

exceptions.FailedToDeleteInstance – if failed to terminate

ocs_ci.utility.aws.update_config_from_s3(bucket_name='ocs-ci-data', filename='auth.yaml')

Get the config file that has secrets/configs from the S3 and update the config

Parameters:

• bucket_name (string) – name of the bucket

• filename (string) – name of the file in bucket

Returns:

returns the updated file contents as python dict None: In case the private bucket could not be accessed

Return type:

dict

ocs_ci.utility.aws.wait_for_distribution_status_deployed(dist_id)

Wait for the Distribution to reach the Deployed status.

Parameters:

dist_id (str) – ID of Distribution

Raises:

exceptions.DistributionStatusError – Raised if status is not Deployed

ocs_ci.utility.azure_utils module

Module for interactions with OCP/OCS Cluster on Azure platform level.

class ocs_ci.utility.azure_utils.AZURE(subscription_id=None, tenant_id=None, client_id=None, client_secret=None, cluster_resource_group=None)

Bases: object

Utility wrapper class for Azure OCP cluster. Design of the class follows similar AWS class.

az_login()

property cluster_resource_group

Azure resource group of the OCP cluster. This group is created by openshift-installer during OCP deployment.

If the value is not yet available and it’s not specified anywhere, it returns None.

property compute_client

Property for Azure vm resource

Returns:

ComputeManagementClient instance for managing Azure vm resource

create_noobaa_managed_identity(cluster_name, resource_group, subscription_id)

Create an Azure user-assigned managed identity for NooBaa with federated credentials for workload identity federation (STS).

Creates:

1. A user-assigned managed identity named ‘{cluster_name}-noobaa-mi’

2. Role assignments: ‘Storage Account Contributor’ and ‘Storage Blob Data Contributor’ scoped to the cluster resource group

3. Federated credentials for NooBaa service accounts (noobaa, noobaa-core, noobaa-endpoint) linked to the cluster’s OIDC issuer

Assumes the caller has already authenticated via az_login().

Parameters:

• cluster_name (str) – Name of the OCP cluster

• resource_group (str) – Azure resource group name

• subscription_id (str) – Azure subscription ID

Returns:

The client ID of the created managed identity

Return type:

str

property credentials

Property for azure service principle credentials used to authenticate the client.

delete_noobaa_managed_identity(cluster_name, resource_group, subscription_id)

Delete the NooBaa Azure managed identity created for STS deployments.

Deleting the MI cascades to its federated credentials, but role assignments are NOT auto-deleted — they become orphaned. This method explicitly removes role assignments before deleting the identity.

Safe to call if the identity doesn’t exist (e.g., partial teardown). Assumes the caller has already authenticated via az_login().

Parameters:

• cluster_name (str) – Name of the OCP cluster

• resource_group (str) – Azure resource group name

• subscription_id (str) – Azure subscription ID

detach_volume(volume, node, timeout=120)

Detach volume if attached

Parameters:

• volume (disk) – disk object required to delete a volume

• node (OCS) – The OCS object representing the node

• timeout (int) – Timeout in seconds for API calls

get_data_volumes(deviceset_pvs)

Get the instance data disk objects

Parameters:

deviceset_pvs (list) – PVC objects of the deviceset PVs

Returns:

Azure Vm disk objects

Return type:

list

get_disk_state(volume_name)

Get the state of the disk

Parameters:

volume_name (str) – Name of the volume/disk

Returns:

Azure Vm disk state

Return type:

str

get_node_by_attached_volume(volume)

Get the Azure Vm instance that has the volume attached to

Parameters:

volume (Disk) – The disk object to get the Azure Vm according to

Returns:

An Azure Vm instance

Return type:

vm

get_storage_account_properties(storage_account_name)

Get the properties of the storage account whose name is passed.

Parameters:

storage_account_name (str) – Name of the storage account

Returns:

Properties of the storage account in string format.

Return type:

str

get_storage_accounts()

Get list of storage accounts in azure resource group

Returns:

list of Azure storage accounts

Return type:

list

get_storage_accounts_names()

Get list of names of storage accounts in azure resource group

Returns:

list of Azure storage accounts name

Return type:

list

get_vm_instance(vm_name)

Get instance of Azure vm Instance

Parameters:

vm_name (str) – The name of the Azure instance to get

Returns:

instance of Azure vm instance resource

Return type:

vm

get_vm_names()

Get list of vms in azure resource group

Returns:

list of Azure vm names

Return type:

(list)

get_vm_power_status(vm_name)

Get the power status of VM

Parameters:

vm_name (str) – Azure VM name

Returns :

str: Power status of Azure VM

property resource_client

Azure ResourceManagementClient instance

restart_vm_instances(vm_names)

Restart Azure vm instances

Parameters:

vm_names (list) – Names of azure vm instances

restart_vm_instances_by_stop_and_start(vm_names, force=False)

Stop and Start Azure vm instances

Parameters:

• vm_names (list) – Names of azure vm instances

• force (bool) – True for non-graceful VM shutdown, False for graceful VM shutdown

set_auth_env_vars()

Set environment variables containing auth information.

start_vm_instances(vm_names)

Start Azure vm instances

Parameters:

vm_names (list) – Names of azure vm instances

stop_vm_instances(vm_names, force=False)

Stop Azure vm instances

Parameters:

• vm_names (list) – Names of azure vm instances

• force (bool) – True for non-graceful VM shutdown, False for graceful VM shutdown

property storage_client

Azure Stroage Management Client instance

class ocs_ci.utility.azure_utils.AzureAroUtil(subscription_id=None, tenant_id=None, client_id=None, client_secret=None, cluster_resource_group=None)

Bases: AZURE

Utility wrapper class for Azure ARO OCP cluster.

check_cluster_response_ok(maximum_attempts=150, successful_connections_in_row=20, insecure=False)

Check if cluster response OK several times based on parameters of the function.

Parameters:

• maximum_attempts (int) – maximum attempts to check cluster is OK

• successful_connections_in_row (int) – how many times the cluster should response OK

• insecure (bool) – if True, it will add –insecure-skip-tls-verify

create_aro_service_principal(cluster_name)

Create a service principal for ARO cluster.

Parameters:

cluster_name (str) – Cluster name.

Returns:

(client_id, client_secret) of the created service principal

Return type:

tuple

create_cluster(cluster_name)

Create OCP cluster.

Parameters:

cluster_name (str) – Cluster name.

Raises:

UnexpectedBehaviour – in the case, the cluster is not installed successfully.

create_network(cluster_name=None)

Create network related stuff for the cluster. Each cluster gets its own VNET with cluster-specific name to allow multiple deployments.

Parameters:

cluster_name (str) – Name of the cluster. If not provided, will use cluster_name from ENV_DATA.

delete_aro_service_principal(cluster_name)

Delete the service principal created for ARO cluster.

Parameters:

cluster_name (str) – Cluster name.

delete_azure_ad_app(cluster_name)

Delete the Azure AD application created by ARO deployment.

When ‘az aro create’ is executed, it automatically creates an Azure AD application named ‘aro-{cluster_name}’. This method cleans up that application.

Parameters:

cluster_name (str) – Cluster name

delete_dns_records(cluster_name, resource_group, base_domain)

Delete DNS records for Azure ARO cluster

Parameters:

• cluster_name (str) – Cluster name.

• resource_group (str) – Base resource group

• base_domain (str) – Base domain for the ARO Cluster

destroy_cluster(cluster_name, resource_group)

Destroy the cluster in Azure ARO.

Performs cleanup in the following order: 1. Delete DNS records 2. Delete ARO cluster 3. Delete cluster resource group (automatically done by az aro delete) 4. Delete VNET 5. Delete Azure AD application

All operations are idempotent and will not fail if resources don’t exist.

Parameters:

• cluster_name (str) – Cluster name

• resource_group (str) – Azure resource group

get_aro_ocp_version()

Get OCP version available in Azure ARO.

Returns:

version of ARO OCP currently available, matching the version from config.

Return type:

str

Raises:

UnsupportedPlatformVersionError – In case the version is not supported yet available for the Azure ARO.

get_cluster_details(cluster_name)

Returns info about the cluster which is taken from the az command.

Parameters:

cluster_name (str) – Cluster name.

Returns:

Cluster details

Return type:

dict

get_kubeconfig(cluster_name, resource_group, path=None)

Export kubeconfig to provided path.

Parameters:

• cluster_name (str) – Cluster name or ID.

• resource_group (str) – Base resource group

• path (str) – Path where to create kubeconfig file.

set_dns_records(cluster_name, resource_group, base_domain)

Set DNS records for Azure ARO cluster.

Parameters:

• cluster_name (str) – Cluster name.

• resource_group (str) – Base resource group

• base_domain (str) – Base domain for the ARO Cluster

update_kubeconfig_with_ocs_ca()

Update kubeconfig with OCS QE CA certificate from ocs-ca-bundle configmap.

This extracts the CA certificate from the cluster’s ocs-ca-bundle configmap (created by configure_ingress_and_api_certificates) and embeds it in the kubeconfig to enable TLS verification.

write_kubeadmin_password(cluster_name, resource_group)

Get kubeadmin password for cluster

ocs_ci.utility.azure_utils.azure_storageaccount_check()

Testing that Azure storage account, post deployment.

Testing for property ‘allow_blob_public_access’ to be ‘false’

ocs_ci.utility.azure_utils.load_cluster_resource_group()

Read terraform tfvars.json file created by openshift-installer in a cluster dir to get azure resource group of an OCP cluster. All Azure resources of the cluster are placed in this group.

Returns:

string with resource group name

Raises:

TerrafromFileNotFoundException – When the terraform tfvars file is not found

ocs_ci.utility.azure_utils.load_service_principal_dict(filepath='/home/docs/.azure/osServicePrincipal.json')

Load Azure Service Principal from osServicePrincipal.json file and parse it into a dictionary.

Parameters:

filepath (str) – path of the

Returns:

dictionary with the service principal details (3 IDs and 1 secret)

ocs_ci.utility.baremetal module

class ocs_ci.utility.baremetal.BAREMETAL

Bases: object

wrapper for Baremetal

get_ipmi_ctx(host, user, password)

Function to get ipmi handler :param host: Host mgmt address :type host: str :param user: User Name for accessing mgmt console :type user: str :param password: Password for accessing mgmt console :type password: str

Returns (object): ipmi handler

get_nodes_ipmi_ctx(baremetal_machine)

Get Node Ipmi handler

Parameters:

baremetal_machine – BM objects

get_power_status(ipmi_ctx)

Get BM Power status

Parameters:

ipmi_ctx (object) – Ipmi host handler

Returns: (bool): bm power status

restart_baremetal_machines(baremetal_machine, force=True)

Restart Baremetal Machines

Parameters:

• baremetal_machine (list) – BM objects

• force (bool) – True for BM ungraceful power off, False for graceful BM shutdown

start_baremetal_machines(baremetal_machine, wait=True)

Start Baremetal Machines

Parameters:

• baremetal_machine (list) – BM objects

• wait (bool) – Wait for BMs to start

start_baremetal_machines_with_ipmi_ctx(ipmi_ctxs, wait=True)

Start Baremetal Machines using Ipmi ctx

Parameters:

• ipmi_ctxs (list) – List of BM ipmi_ctx

• wait (bool) – Wait for BMs to start

stop_baremetal_machines(baremetal_machine, force=True)

Stop Baremetal Machines

Parameters:

• baremetal_machine (list) – BM objects

• force (bool) – True for BM ungraceful power off, False for graceful BM shutdown

Raises:

UnexpectedBehaviour – If baremetal machine is still up

verify_machine_is_down(node)

Verifiy Baremetal machine is completely power off

Parameters:

node (object) – Node objects

Returns:

True if machine is down, False otherwise

Return type:

bool

ocs_ci.utility.baremetal.update_uefi_boot_order(first_option_string=None, ocp_node=None, host=None, ignore_error=False)

Update UEFI boot order (using efibootmgr) to set option containing first_option_string to first place. Depending on the ocp_node and host parameters, it tries to access the node via oc debug command (if ocp_node parameter is provided) or via ssh (as core user) if host (hostname or IP) is provided. Parameter ocp_node have precedence before host parameter, which means that if both parameters are provided, it will try to use only the oc debug access.

Parameters:

• first_option_string (str) – String identifying the boot option which should be set to first place

• ocp_node (str) – OCP Node name used to access the server via oc debug command

• host (dict) – configuration of the server to make ssh connection or None the dict could contain following keys: host, user and private_key or password

• ignore_error (bool) – If True, do not raise an exception if the command fails for some reason.

Raises:

UnexpectedBehaviour – If parsing of output from efibootmgr command fails

ocs_ci.utility.bootstrap module

ocs_ci.utility.bootstrap.gather_bootstrap()

Gather debugging data for a failing-to-bootstrap control plane. Data is placed in the gather_bootstrap directory under the log directory.

Raises:

NodeNotFoundError – If we are unable to retrieve the IP of any master nodes

ocs_ci.utility.bootstrap.get_gather_bootstrap_node_data()

Retrieve node IPs required by the gather bootstrap command

Raises:

UnsupportedPlatformError – If we do not support gathering bootstrap data for the configured provider

Returns:

Public IP of the bootstrap node and Private IPs of master nodes

Return type:

dict

ocs_ci.utility.bootstrap.get_node_data_aws()

Retrieve bootstrap public IP and master node private IPs running in aws

Raises:

NodeNotFoundError – If we are unable to find the bootstrap node or IP

Returns:

bootstrap and master node IP data

Return type:

dict

ocs_ci.utility.bootstrap.get_node_data_vsphere()

Retrieve bootstrap public IP and master node private IPs running in vsphere

Raises:

NodeNotFoundError – If we are unable to find the bootstrap node or IP

Returns:

bootstrap and master node IP data

Return type:

dict

ocs_ci.utility.cco module

Cloud Credential Operator utility functions

ocs_ci.utility.cco.configure_cloud_credential_operator()

Extract and Prepare the CCO utility (ccoctl) binary. This utility allows us to create and manage cloud credentials from outside of the cluster while in manual mode.

ocs_ci.utility.cco.create_manifests(openshift_installer, output_dir)

Create manifests.

Parameters:

• openshift_installer (str) – Path to the openshift installer

• output_dir (str) – Path to the output directory

ocs_ci.utility.cco.create_service_id(cluster_name, cluster_path, credentials_requests_dir)

Create the Service ID.

Parameters:

• cluster_name (str) – Name of the cluster

• cluster_path (str) – Path to the cluster directory

• credentials_requests_dir (str) – Path to the credentials requests directory

ocs_ci.utility.cco.delete_oidc_resource_group(name, region, subscription_id)

Delete the Azure resources that ccoctl created.

Parameters:

• name (str) – Name used to tag any created cloud resources

• region (str) – Region to create cloud resources

• subscription_id (str) – Service Principal Subscription ID

ocs_ci.utility.cco.delete_service_id(cluster_name, credentials_requests_dir)

Delete the Service ID.

Parameters:

• cluster_name (str) – Name of the cluster

• credentials_requests_dir (str) – Path to the credentials requests directory

ocs_ci.utility.cco.extract_ccoctl_binary(cco_image, pull_secret_path)

Extract the ccoctl binary from the CCO container image within the OCP release image.

Parameters:

• cco_image (str) – Release image from the openshift installer

• pull_secret_path (str) – Path to the pull secret

ocs_ci.utility.cco.extract_credentials_requests(release_image, install_config, pull_secret, credentials_requests_dir)

Extract the CredentialsRequests (AWS and Azure STS variant).

Parameters:

• release_image (str) – Release image from the openshift installer

• install_config (str) – Location of the install-config.yaml

• credentials_requests_dir (str) – Path to the CredentialsRequests directory

ocs_ci.utility.cco.extract_credentials_requests_ibmcloud(release_image, credentials_requests_dir, pull_secret_path)

Extract the CredentialsRequests (IBM Cloud variant).

Parameters:

• release_image (str) – Release image from the openshift installer

• credentials_requests_dir (str) – Path to the CredentialsRequests directory

• pull_secret_path (str) – Path to the pull secret

ocs_ci.utility.cco.get_cco_container_image(release_image, pull_secret_path)

Obtain the CCO container image from the OCP release image.

Parameters:

• release_image (str) – Release image from the openshift installer

• pull_secret_path (str) – Path to the pull secret

Returns:

ocs_ci.utility.cco.process_credentials_requests_aws(name, aws_region, credentials_requests_dir, output_dir)

Process all CredentialsRequest objects.

Parameters:

• name (str) – Name used to tag any created cloud resources

• aws_region (str) – Region to create cloud resources

• credentials_requests_dir (str) – Path to the CredentialsRequest directory

• output_dir (str) – Path to the output directory

ocs_ci.utility.cco.process_credentials_requests_azure(name, azure_region, credentials_requests_dir, output_dir, subscription_id, dns_zone_group_name, tenant_id)

Process all CredentialsRequest objects.

Parameters:

• name (str) – Name used to tag any created cloud resources

• azure_region (str) – Region to create cloud resources

• credentials_requests_dir (str) – Path to the CredentialsRequest directory

• output_dir (str) – Path to the output directory

• subscription_id (str) – Service Principal Subscription ID

• dns_zone_group_name (str) – Name of the DNS Zone

• tenant_id (str) – Service Principal Tenant ID

ocs_ci.utility.cco.set_credentials_mode_manual(install_config)

Set credentialsMode to Manual in the install-config.yaml

ocs_ci.utility.cco.set_resource_group_name(install_config, name)

Set resourceGroupName to Manual in the install-config.yaml for Azure deployments.

Parameters:

• install_config (str) – Path to the install-config.yaml

• name (str) – Name of the Resource Group

ocs_ci.utility.connection module

Module that connects to remote server and execute operations on remote server

class ocs_ci.utility.connection.Connection(host, user=None, private_key=None, password=None, stdout=False, jump_host=None)

Bases: object

A class that connects to remote server

exec_cmd(cmd, secrets=None)

Executes command on server

Parameters:

• cmd (str) – Command to run on server

• secrets (list) – A list of secrets to be masked with asterisks This kwarg is popped in order to not interfere with subprocess.run(**kwargs)

Returns:

tuple which contains command return code, output and error

Return type:

tuple

upload_file(localpath, remotepath)

Upload a file to remote server

Parameters:

• localpath (str) – Local file to upload

• remotepath (str) – Target path on the remote server. Filename should be included

ocs_ci.utility.csr module

ocs_ci.utility.csr.approve_csrs(pending_csrs)

Approves the CSRs

Parameters:

csrs (list) – List of CSRs

ocs_ci.utility.csr.approve_pending_csr(expected=None)

After node addition CSR could be in pending state, we have to approve it.

Parameters:

expected (int) – Expected number of CSRs. By default, it will approve all the pending CSRs if exists.

Raises:

• exceptions.PendingCSRException –

• exceptions.TimeoutExpiredError –

ocs_ci.utility.csr.check_no_pending_csr()

Check whether we have any pending CSRs.

Raises:

exceptions.PendingCSRException –

ocs_ci.utility.csr.get_csr_resource()

Retrieve the latest CSR data

Returns:

CSR data

Return type:

ocp.OCP

ocs_ci.utility.csr.get_nodes_csr()

Fetches the CSRs

Returns:

Dictionary with node as keys and CSRs as values

e.g:{

‘control-plane-1’: [‘csr-6hx8z’], ‘compute-2’: [‘csr-blz2n’], ‘compute-0’: [‘csr-p46bz’], ‘control-plane-2’: [‘csr-tzhn5’], ‘control-plane-0’: [‘csr-wm4g5’] }

Return type:

dict

ocs_ci.utility.csr.get_pending_csr()

Gets the pending CSRs

Returns:

list of pending CSRs

Return type:

list

ocs_ci.utility.csr.wait_for_all_nodes_csr_and_approve(timeout=900, sleep=10, expected_node_num=None, ignore_existing_csr=None)

Wait for CSR to generate for nodes

Parameters:

• timeout (int) – Time in seconds to wait

• sleep (int) – Sampling time in seconds

• expected_node_num (int) – Number of nodes to verify CSR is generated

• ignore_existing_csr (dct) – Existing CSR to ignore

• e.g –

  { ‘compute-1’: [‘csr-64vkw’] }

Returns:

True if all nodes are generated CSR

Return type:

bool

Raises:

TimeoutExpiredError – in case CSR not found

ocs_ci.utility.decorators module

ocs_ci.utility.decorators.enable_high_recovery(func)

Decorator to temporarily boost Ceph recovery performance during the execution of the wrapped function by applying both Ceph and mClock high-recovery profiles.

This is useful for operations such as OSD replacement or data rebalancing where faster recovery is desired at the cost of reduced client I/O bandwidth.

Behavior: - Sets the Ceph recovery profile to ‘high_recovery_ops’ - Sets the mClock profile to ‘high_client_ops’ - Restores both profiles to their original or default state after function execution

If the ODF CLI runner is not available or the current profile cannot be determined, the wrapped function is executed without making any changes.

Returns:

The result of the wrapped function.

ocs_ci.utility.decorators.enable_high_recovery_during_rebalance_flag(func)

Decorator that applies ‘enable_high_recovery’ only if the ‘enable_high_recovery_during_rebalance’ flag is set to True in the test configuration. Otherwise, the function runs as-is.

Returns:

The result of the wrapped function.

ocs_ci.utility.decorators.switch_to_client_for_function(func, client_index=0)

A decorator for switching to the client cluster for the function execution. After the function execution, it switches back to the original index.

Example of usage: Suppose we have the function definition ‘def wait_for_storage_client_connected(timeout=180, sleep=10)’. Here are three examples of usage: 1. switch_to_client_for_function(wait_for_storage_client_connected)(timeout=30) 2. switch_to_client_for_function(wait_for_storage_client_connected, client_index=1)() 2. switch_to_client_for_function(wait_for_storage_client_connected, client_index=1)(timeout=30, sleep=5)

Parameters:

• func (function) – The function we want to decorate

• client_index (int) – The client cluster index to switch. The default value is 0.

ocs_ci.utility.decorators.switch_to_default_cluster_index_at_last(func)

A decorator for switching to the default cluster index after the function execution. This decorator will primarily be used in the ‘teardown’ and ‘finalizer’ methods when we want to make sure that the next test will start with the default cluster index.

Parameters:

func (function) – The function we want to decorate

ocs_ci.utility.decorators.switch_to_orig_index_at_last(func)

A decorator for switching to the original index after the function execution

Parameters:

func (function) – The function we want to decorate

ocs_ci.utility.decorators.switch_to_provider_for_function(func)

A decorator for switching to the provider cluster for the function execution. After the function execution, it switches back to the original index.

Parameters:

func (function) – The function we want to decorate

ocs_ci.utility.deployment module

Utility functions that are used as a part of OCP or OCS deployments

ocs_ci.utility.deployment.add_mc_partitioned_disk_on_workers_to_ocp_deployment(disk)

Add Machine Config for partitioned disk on worker nodes to OCP deployment

Parameters:

disk (str) – path to root disk where the additional partition should be created common for all worker nodes

ocs_ci.utility.deployment.apply_idms_via_worker_filesystem(image_digest_mirrors, conf_filename=None)

Write CRI-O registry mirror configuration to worker nodes via a privileged DaemonSet that mounts the host filesystem at /host.

Used on ROSA HCP where ImageDigestMirrorSet cannot be applied via oc apply (blocked by ValidatingAdmissionPolicy).

Parameters:

• image_digest_mirrors (list) – list of dicts with ‘source’ and ‘mirrors’ keys, matching the imageDigestMirrors schema from an ImageDigestMirrorSet.

• conf_filename (str) – filename to write inside registries.conf.d. Defaults to ROSA_HCP_KONFLUX_MIRROR_CONF constant.

ocs_ci.utility.deployment.create_external_secret(ocs_version=None, apply=False)

Creates secret data for external cluster

Parameters:

• ocs_version (str) – OCS version

• apply (bool) – True if want to use apply instead of create command

ocs_ci.utility.deployment.create_openshift_install_log_file(cluster_path, console_url)

Workaround. Create .openshift_install.log file containing URL to OpenShift console. It is used by our CI jobs to show the console URL in build description.

Parameters:

• cluster_path (str) – The path to the cluster directory.

• console_url (str) – The address of the OpenShift cluster management-console

ocs_ci.utility.deployment.deploy_roks_icsp_daemonset()

Deploy the roks-icsp privileged DaemonSet on ROSA HCP worker nodes.

The DaemonSet (quay.io/cicdtest/roks-enabler:rosa) mounts the host filesystem at /host and is used to write CRI-O mirror configuration directly to /host/etc/containers/registries.conf.d/ on each worker, bypassing the ImageDigestMirrorSet API which is blocked on ROSA HCP by ValidatingAdmissionPolicy.

Also creates a fake machineconfigs.machineconfiguration.openshift.io CRD which is absent on ROSA HCP hosted clusters.

Source: https://github.com/xcliu-ca/rosa-icsp-gps/blob/main/enabler.sh

ocs_ci.utility.deployment.get_and_apply_idms_from_catalog(image, apply=True, insecure=False)

Get IDMS from catalog image (if exists) and apply it on the cluster (if requested).

Parameters:

• image (str) – catalog image of ocs registry.

• apply (bool) – controls if the IDMS should be applied or not (default: true)

• insecure (bool) – If True, it allows push and pull operations to registries to be made over HTTP

Returns:

path to the idms.yaml file or empty string, if idms not available

in the catalog image

Return type:

str

ocs_ci.utility.deployment.get_cluster_prefix(cluster_name, special_rules)

Parse out the “prefix” of a cluster name. Note this is not the same thing as the CLUSTER_PREFIX in jenkins. In fact we will parse that value out. This “cluster prefix” is used to check cloud providers to see if a particular user already has a cluster created. This is to stop people from using too many cloud resources at one time.

Parameters:

• cluster_name (str) – name of the cluster

• special_rules (dict) – dictionary containing special prefix rules that allow clusters to remain alive longer than our default value

Returns:

cluster name prefix

Return type:

str

ocs_ci.utility.deployment.get_ocp_ga_version(channel)

Retrieve the latest GA version for

Parameters:

channel (str) – the OCP version channel to retrieve GA version for

Returns:

latest GA version for the provided channel.

An empty string is returned if no version exists.

Return type:

str

ocs_ci.utility.deployment.get_ocp_release_image()

Get the url of ocp release image * from DEPLOYMENT[“custom_ocp_image”] or * from openshift-install version command output

Returns:

Release image of the openshift installer

Return type:

str

ocs_ci.utility.deployment.get_ocp_release_image_from_installer()

Retrieve release image using the openshift installer.

Returns:

Release image of the openshift installer

Return type:

str

ocs_ci.utility.deployment.workaround_mark_disks_as_ssd()

This function creates MachineConfig defining new service workaround-ssd, which configures all disks as SSD (not rotational). This is useful for example on some Bare metal servers where are SSD disks not properly recognized as SSD, because of wrong RAID controller configuration or issue.

ocs_ci.utility.deployment_openshift_logging module

This module deploys the openshift-logging on the cluster lokistack stack

ocs_ci.utility.deployment_openshift_logging.create_UI_Plugin(yaml_file, resource_name)

Creates UI-Plugin for cluster obervability operator

Parameters:

• yaml_file – path to yaml file to create UIPlugin

• resource_name – name of the UIPlugin

Returns:

True if plugin is created

successfully, false otherwise

Return type:

bool

ocs_ci.utility.deployment_openshift_logging.create_clusterlogforwarder(yaml_file, skip_resource_exists=False)

Create a ClusterLogForwarder CR to collect logs from nodes and application containers and verifies verify successfull installation of cfr by checking required no of pods and pvc running in project

Parameters:

• yaml_file – path tp where yaml for clusterlogforwader exists

• skip_resource_exists – Skip the resource creation if it already exists

ocs_ci.utility.deployment_openshift_logging.create_clusterlogging_operator_group(yaml_file, skip_resource_exists=False)

Creation of operator-group for clusterlogging operator.

Parameters:

• yaml_file (str) – Path to yaml file to create operator group for cluster-logging operator

• skip_resource_exists – Skip the resource creation if it already exists

Returns:

True if operator group for cluster-logging is created

successfully, false otherwise

Return type:

bool

Example

create_clusterlogging_operator_group(yaml_file=constants.CL_OG_YAML)

ocs_ci.utility.deployment_openshift_logging.create_clusterobservability_operator_group(yaml_file, resource_name, skip_resource_exists=False)

Creation of operator-group for cluster obervability operator :param yaml_file: Path to yaml file to create operator group for :type yaml_file: str :param resource_name: Name of the operator group to create for :type resource_name: str :param skip_resource_exists: Skip the resource creation if it already exists

Returns:

True if operator group is created

successfully, false otherwise

Return type:

bool

ocs_ci.utility.deployment_openshift_logging.create_lokistack(yaml_file, skip_resource_exists=False)

creates lokistack (logging store) for storing logs from nodes and application containers

Parameters:

• yaml_file (str) – path to where lokistack yaml exists

• skip_resource_exists – Skip the resource creation if it already exists

ocs_ci.utility.deployment_openshift_logging.create_lokistack_operator_group(yaml_file, resource_name, skip_resource_exists=False)

Creation of operator-group for lokitsack operator

Parameters:

• yaml_file (str) – Path to yaml file to create operator group for elastic-search

• resource_name (str) – Name of the operator group to create for elastic-search

• skip_resource_exists – Skip the resource creation if it already exists

Returns:

True if operator group for lokistack is created

successfully, false otherwise

Return type:

bool

Example:

create_lokistack_operator_group(
    constants.EO_OG_YAML, 'openshift-operators-redhat'
)

ocs_ci.utility.deployment_openshift_logging.create_namespace(yaml_file, skip_resource_exists=False)

Creation of namespace “openshift-operators-redhat” for lokistack-operator and “openshift-logging” for ClusterLogging-operator

Parameters:

• yaml_file (str) – Path to yaml file to create namespace

• skip_resource_exists – Skip the namespace creation if it already exists

Example

create_namespace(yaml_file=constants.EO_NAMESPACE_YAML)

ocs_ci.utility.deployment_openshift_logging.get_cluster_observability_subscription()

Fetches subscription for cluster obervability operator

Returns:

True if subscription is created

successfully, false otherwise

Return type:

bool

ocs_ci.utility.deployment_openshift_logging.get_clusterlogging_subscription()

Creation of subscription for clusterlogging to subscribe a namespace to an operator

Returns:

True if subscription exists

Return type:

bool

Example

cl_create_subscription(yaml_file=constants.CL_SUB_YAML)

ocs_ci.utility.deployment_openshift_logging.get_lokistack_subscription()

Creation of subscription for lokistack to subscribe a namespace to an operator

Returns:

Subscription exists or not

Return type:

bool

ocs_ci.utility.deployment_openshift_logging.get_obc()

Checking for successful creation of OBC for providing s3 object storage for lokistack.

Returns:

The return value. True for success if obc created, False otherwise.

Return type:

bool

ocs_ci.utility.deployment_openshift_logging.get_secret_to_lokistack()

check if secret is created successfully that will contains endpoint and credential details for s3 bucket used by lokistack

Returns:

True if secret created successfully

Return type:

bool

ocs_ci.utility.deployment_openshift_logging.get_storage_class_for_logging()

Determine the appropriate storage class for logging based on platform type.

Returns:

Storage class name to use for logging OBC

Return type:

str

ocs_ci.utility.deployment_openshift_logging.install_logging(skip_resource_exists=False)

Install openshift-logging with LokiStack

Parameters:

skip_resource_exists (bool) – Skip resource creation if it already exists (useful for ROSA HCP deployments)

ocs_ci.utility.deployment_openshift_logging.setup_sa_permissions()

Assign the necessary permissions to the service account for the collector to be able to collect and forward logs

the collector is provided permissions to collect logs from infrastructure application and audit logs.

ocs_ci.utility.environment_check module

Util for environment check before and after test to compare and find stale leftovers

ocs_ci.utility.environment_check.assign_get_values(env_status_dict, key, kind=None, exclude_labels=None, exclude_job_owned_pods=True)

Assigning kind status into env_status_dict

Parameters:

• env_status_dict (dict) – Dictionary which is copy.deepcopy(ENV_STATUS_DICT)

• key (str) – Name of the resource

• kind (OCP obj) – OCP object for a resource

• exclude_labels (list) – App labels to ignore leftovers

• exclude_job_owned_pods (bool) – If True, exclude pods owned by Jobs

ocs_ci.utility.environment_check.compare_dicts(before, after)

Comparing 2 dicts and providing diff list of [added items, removed items]

Parameters:

• before (dict) – Dictionary before execution

• after (dict) – Dictionary after execution

Returns:

List of 2 lists - (‘added’ and ‘removed’ are lists) None: If both parameters are None

Return type:

list

ocs_ci.utility.environment_check.get_environment_status(env_dict, exclude_labels=None, exclude_job_owned_pods=True)

Get the environment status per kind in KINDS and save it in a dictionary

Parameters:

• env_dict (dict) – Dictionary that is a copy.deepcopy(ENV_STATUS_DICT)

• exclude_labels (list) – App labels to ignore leftovers

• exclude_job_owned_pods (bool) – If True, exclude pods owned by Jobs

ocs_ci.utility.environment_check.get_status_after_execution(exclude_labels=None, exclude_job_owned_pods=True)

Set the environment status and assign it into ENV_STATUS_PRE dictionary. In addition compare the dict before the execution and after using DeepDiff

Parameters:

• exclude_labels (list) – App labels to ignore leftovers

• exclude_job_owned_pods (bool) – If True, exclude pods owned by Jobs

Raises:

ResourceLeftoversException – In case there are leftovers in the environment after the execution

ocs_ci.utility.environment_check.get_status_before_execution(exclude_labels=None, exclude_job_owned_pods=True)

Set the environment status and assign it into ENV_STATUS_PRE dictionary

Parameters:

• exclude_labels (list) – App labels to ignore leftovers

• exclude_job_owned_pods (bool) – If True, exclude pods owned by Jobs

ocs_ci.utility.flexy module

Utility functions related to OCP deployment done via Flexy

ocs_ci.utility.flexy.configure_allowed_domains_in_proxy()

Configure squid proxy server - add domains which needs to be accessible from disconnected cluster for test execution.

ocs_ci.utility.flexy.load_cluster_info(cluster_config=None)

Parse cluster_info.json and cluster_info.yaml files generated by Flexy and load relevant configuration.

Parameters:

cluster_config (MultiClusterConfig) – config object of a specific cluster in case of multiclsuter scenario, if its single cluster we will go ahead with the config

ocs_ci.utility.gcp module

Module for interactions with OCP/OCS Cluster on Google Cloud platform level.

It’s using libcloud module as much as possible, but if that is not feasible, we can use module from Google Cloud python libraries as well. This is not the case so far.

class ocs_ci.utility.gcp.GoogleCloud(project_id=None)

Bases: object

This is a wrapper class for GoogleCloud

base_instance_operation(operation_name, zone, instance_name)

The base method for performing the instance operations: ‘stop’, ‘start’, ‘restart’, ‘terminate’.

Parameters:

• operation_name (str) – The instance operation name(‘stop’, ‘start’, etc.)

• zone – The instance zone

• instance_name – The instance name

Returns:

The result of the execution made by the instance operation

Return type:

dict

base_instances_operation(operation_name, instance_names, wait=True)

The base method for performing the instances operations: ‘stop’, ‘start’, ‘restart’, ‘terminate’.

Parameters:

• operation_name (str) – The operation name to perform on the instances

• instance_names (list) – The instance names

• wait (bool) – If True, will wait for the operation to complete. False, otherwise

Raises:

OperationFailedToCompleteException – In case that not all the operations completed successfully

get_instance_zone_dict()

Get the instance name per instance zone dictionary for all the instances

Returns:

The instance name per instance zone dictionary for all the instances

Return type:

dict

get_instances_zones(instance_names)

Get the zones of the given instance names

Parameters:

instance_names – The instance names to get their zones

Returns:

The zones of the given instance names

Return type:

list

get_operation_data(zone, operation_id)

Get the operation data of a given operation id. (For example after stopping an instance, get the data of the stop operation id)

Parameters:

• zone (str) – The zone of the operation id

• operation_id (str) – The operation id

Returns:

The operation data of a given operation id

Return type:

dict

get_operations_data(zone_operation_id_dict)

Get the operations data for the given operation ids

Parameters:

zone_operation_id_dict – A dictionary of the operation id zone per operation id

Returns:

The operations data for the given operation ids

Return type:

list

restart_instances(instance_names, wait=True)

Restart instances. This is a hard reset - the instance does not do a graceful shutdown

Parameters:

• instance_names (list) – The instance names to restart

• wait (bool) – If True, wait for the instances to be ready. False, otherwise.

Raises:

OperationFailedToCompleteException – If wait is True, and not all the operations completed successfully

restart_instances_by_stop_and_start(instance_names, wait=True)

Restart instances by stop and start

Parameters:

• instance_names (list) – The instance names to restart

• wait (bool) – If True, wait for the instances to be ready. False, otherwise.

Raises:

OperationFailedToCompleteException – If wait is True, and not all the operations completed successfully

start_instances(instance_names, wait=True)

Start instances

Parameters:

• instance_names (list) – The instance names to start

• wait (bool) – If True, wait for the instances to be ready. False, otherwise.

Raises:

OperationFailedToCompleteException – If wait is True, and not all the operations completed successfully

stop_instances(instance_names, wait=True)

Stop instances

Parameters:

• instance_names (list) – The instance names to stop

• wait (bool) – If True, wait for the instances to stop. False, otherwise.

Raises:

OperationFailedToCompleteException – If wait is True, and not all the operations completed successfully

terminate_instances(instance_names, wait=True)

Terminate instances

Parameters:

• instance_names (list) – The instance names to terminate

• wait (bool) – If True, wait for the instances to terminate. False, otherwise.

Raises:

OperationFailedToCompleteException – If wait is True, and not all the operations completed successfully

wait_for_operations_to_complete(zone_operation_id_dict, timeout=300, sleep=10)

Wait for the operations with the given operation IDs to complete

Parameters:

• zone_operation_id_dict (dict) – The operation zone per operation id dictionary

• timeout (int) – Time in seconds to wait for the operation to complete

• sleep (int) – Time in seconds to wait between iterations

Raises:

OperationFailedToCompleteException – In case that not all the operations completed successfully

class ocs_ci.utility.gcp.GoogleCloudUtil(region_name=None)

Bases: object

Utility wrapper class for Google Cloud OCP cluster. Design of the class follows similar AWS and Azure class.

property compute_driver

Compute Driver instance for GCP.

property service_account

Dictionary with GCP service account details, which contains authentication keys and cluster details loaded from Service Account Key file.

ocs_ci.utility.gcp.SERVICE_ACCOUNT_KEY_FILEPATH = '/home/docs/.gcp/osServiceAccount.json'

absolute filepath of json file with service account key

This is json key file of sg-serv-account service account, which has full admin rights in given GCP project. The same key file is used by openshift installer during OCP installation to create all cluster resources from virtual machines to hostnames. Modules from ocs-ci are using the same key to get full cluster access as well.

For more details, see GCP documentation on ServiceAccountKey resource

Type:

str

ocs_ci.utility.gcp.load_service_account_key_dict(filepath='/home/docs/.gcp/osServiceAccount.json')

Load GCP Service Account key from osServiceAccount.json file and parse it into a dictionary.

Parameters:

filepath (str) – path of the osServiceAccount.json file

Returns:

dictionary with the service account details

ocs_ci.utility.iam_utils module

ocs_ci.utility.iam_utils.generate_random_iam_path(levels=3, length=8)

Generates random path of iam user

Parameters:

• levels (int) – Number of levels (nesting) of the path

• length (int) – Length of each part of path

Returns:

random iam user path

Return type:

str

ocs_ci.utility.iam_utils.get_base_iam_command(mcg_obj)

Builds base iam command, to which “a real command” should be added by the caller

Parameters:

mcg_obj – An MCG class instance

Returns:

base iam command

Return type:

str

Raises:

ValueError – If IAM endpoint is not available (e.g., on Azure ARO)

ocs_ci.utility.iam_utils.get_user_access_keys(mcg_obj, awscli_pod_session, user_name)

Runs user-access-keys command

Parameters:

• mcg_obj – An MCG class instance

• awscli_pod_session (pod) – A pod running the AWSCLI tools

• user_name (str) – Name of the user whose access keys should be returned

Returns:

user-access-keys command result

Return type:

dict

ocs_ci.utility.iam_utils.run_iam_command(mcg_obj, awscli_pod_session, cmd)

Builds base iam command, to which “a real command” should be added by the caller

Parameters:

• mcg_obj – An MCG class instance

• awscli_pod_session (pod) – A pod running the AWSCLI tools

• cmd (str) – A command to run

Returns:

command result

Return type:

dict

ocs_ci.utility.ibmcloud module

Module for interactions with IBM Cloud Cluster.

class ocs_ci.utility.ibmcloud.IBMCloud

Bases: object

Wrapper for Ibm Cloud

attach_volume(volume, node)

Attach volume to node on IBM Cloud.

Parameters:

• volume (str) – volume id.

• node (OCS) – worker node id to attach.

delete_volume_id(volume)

Deletes Volumeid

Parameters:

volume (str) – The volume to be deleted

detach_volume(volume, node=None, delete_from_backend=True)

Detach volume from node on IBM Cloud.

Parameters:

• volume (str) – volume id.

• node (OCS) – worker node id to detach.

• delete_from_backend (bool) – True for deleting the disk from the storage backend, False otherwise

get_data_volumes()

Returns volumes in IBM Cloud for cluster.

Returns:

volumes in IBM Cloud for cluster.

Return type:

list

get_node_by_attached_volume(volume)

Get the node by attached volume on IBM Cloud.

Parameters:

volume (str) – volume id.

Raises:

NodeHasNoAttachedVolume – In case the volume is not attached to node

Returns:

worker id

Return type:

str

get_volume_id()

Returns Volumeid with the name taken from constants

Returns:

volume id if the volume exists otherwise create

new volume

Return type:

str

is_volume_attached(volume)

Check if volume is attached to node or not.

Parameters:

volume (str) – The volume to check for to attached

Returns:

‘True’ if volume is attached otherwise ‘False’

Return type:

bool

restart_nodes(nodes, timeout=900, wait=True)

Reboot the nodes on IBM Cloud.

Parameters:

nodes (list) – The worker node instance

restart_nodes_by_stop_and_start(nodes, force=True)

Reboot the nodes which are not ready on IBM Cloud.

Parameters:

• nodes (list) – The OCS objects of the nodes

• force (bool) – True for force VM stop, False otherwise

start_nodes(nodes, wait=True)

Start nodes on IBM Cloud.

Parameters:

• nodes (list) – The OCS objects of the nodes

• wait (bool) – True for waiting the instances to start, False otherwise

Raises:

ValueError – if the list of nodes is empty

stop_nodes(nodes, wait=True)

Stop nodes on IBM Cloud

Parameters:

• nodes (list) – The OCS objects of the nodes

• wait (bool) – True for waiting the instances to stop, False otherwise

Raises:

ValueError – if the list of nodes is empty

wait_for_volume_attach(volume)

Checks volume is attached to node or not

Parameters:

volume (str) – The volume to wait for to be attached

Returns:

True if the volume has been attached to the

instance, False otherwise

Return type:

bool

class ocs_ci.utility.ibmcloud.IBMCloudIPI

Bases: object

Wrapper for Ibm Cloud IPI

attach_volume(volume, node)

Attach volume to node on IBM Cloud. :param volume: volume id. :type volume: str :param node: worker node object to attach. :type node: OCS

check_node_status(node_name, node_status)

Check the node status in IBM cloud

Parameters:

• node_name (str) – Node name

• node_status (str) – Status of Node Running or Stopped

Returns:

True if status matches else False

Return type:

bool

detach_volume(volume, node=None)

Detach volume from node on IBM Cloud. :param volume: volume id. :type volume: str :param node: worker node object to detach. :type node: OCS

is_volume_attached(volume)

Check if volume is attached to node or not.

Parameters:

volume (str) – The volume to check for to attached

Returns:

‘True’ if volume is attached otherwise ‘False’

Return type:

bool

restart_nodes(nodes, wait=True, timeout=900)

Reboot the nodes on IBM Cloud. :param nodes: The worker node instance :type nodes: list :param wait: Wait for the VMs to stop :type wait: bool :param timeout: Timeout for the command, defaults to 900 seconds. :type timeout: int

restart_nodes_by_stop_and_start(nodes, wait=True, force=True, timeout=300)

Restart nodes by stopping and starting VM in IBM Cloud :param nodes: The OCS objects of the nodes :type nodes: list :param wait: True in case wait for status is needed, False otherwise :type wait: bool :param force: True for force instance stop, False otherwise :type force: bool :param timeout: Timeout for the command, defaults to 300 seconds. :type timeout: int

restart_nodes_by_stop_and_start_force()

Make sure all nodes are up by the end of the test on IBM Cloud.

start_nodes(nodes)

Start the nodes on IBM Cloud :param nodes: The OCS objects of the nodes :type nodes: list

stop_nodes(nodes, force=True, wait=True)

Stop the nodes on IBM Cloud :param nodes: The OCS objects of the nodes :type nodes: list :param force: True for VM ungraceful power off, False for graceful VM shutdown :type force: bool :param wait: Wait for the VMs to stop :type wait: bool

terminate_nodes(nodes, wait=True)

Terminate the Node in IBMCloud :param nodes: The OCS objects of the nodes :type nodes: list :param wait: True in case wait for status is needed, False otherwise :type wait: bool

wait_for_volume_attach(volume)

Checks volume is attached to node or not :param volume: The volume to wait for to be attached :type volume: str

Returns:

True if the volume has been attached to the instance, False otherwise

Return type:

bool

class ocs_ci.utility.ibmcloud.IBMCloudObjectStorage(api_key, service_instance_id, endpoint_url)

Bases: object

IBM Cloud Object Storage (COS) class

delete_bucket(bucket_name)

Delete the bucket

Parameters:

bucket_name (str) – Name of the bucket

delete_objects(bucket_name)

Delete objects in a bucket

Parameters:

bucket_name (str) – Name of the bucket

get_bucket_objects(bucket_name, prefix=None)

Fetches the objects in a bucket

Parameters:

• bucket_name (str) – Name of the bucket

• prefix (str) – Prefix for the objects to fetch

Returns:

List of objects in a bucket

Return type:

list

get_buckets()

Fetches the buckets

Returns:

List of buckets

Return type:

list

ocs_ci.utility.ibmcloud.add_deployment_dependencies()

Adding dependencies for IBM Cloud deployment

ocs_ci.utility.ibmcloud.add_security_group_rule(security_group, direction, protocol, port_min, port_max, **kwargs)

Add security group rule

Parameters:

• security_group (str) – security group ID or Name

• direction (str) – inbound or outbound

• protocol (str) – protocol, e.g. tcp, udp, icmp, all

• port_min (int) – minimum port number

• port_max (int) – maximum port number

• **kwargs – other arguments to be passed to command, e.g. –vpc ID or name of the VPC. It is required to specify only the unique resource by name inside this VPC.

ocs_ci.utility.ibmcloud.assign_floating_ips_to_workers()

Assigns floating IPs to all worker instances that do not already have one.

Returns:

Mapping of worker VSI names to their assigned floating IPs.

Return type:

dict

ocs_ci.utility.ibmcloud.attach_subnet_to_public_gateway(subnet_name, gateway_name, vpc)

Attach subnet to public gateway.

Parameters:

• subnet_name (str) – subnet name to attach to public gateway

• gateway_name (str) – public gateway name

• vpc (str) – VPC name

ocs_ci.utility.ibmcloud.cleanup_policies_and_service_ids(cluster_name, get_infra_id_from_metadata=True)

Cleanup Account Policies and Service IDs for cluster.

Parameters:

• cluster_name (str) – cluster name

• get_infra_id_from_metadata (bool) – if set to true it will try to get infra ID from metadata.json file (Default: True)

ocs_ci.utility.ibmcloud.configure_ingress_load_balancer_security_group()

Add inbound rules for ports 80 and 443 to the default ingress load balancer security group. This is required for IBM Cloud IPI deployments to ensure ingress availability.

The function: 1. Gets the router-default service from openshift-ingress namespace 2. Extracts the load balancer hostname 3. Finds the IBM Cloud load balancer by hostname (filtered by resource group) 4. Gets the security group(s) attached to that load balancer 5. Adds inbound TCP rules for HTTP (80) and HTTPS (443) ports

ocs_ci.utility.ibmcloud.configure_nfs_lb_security_group()

Add an inbound TCP rule for port 2049 (NFS) to the security groups attached to the NFS LoadBalancer on IBM Cloud VPC.

Must be called after the rook-ceph-nfs-my-nfs-load-balancer Service has an ingress address assigned.

ocs_ci.utility.ibmcloud.create_address_prefix(prefix_name, vpc, zone, cidr)

Create address prefix in VPC.

Parameters:

• prefix_name (str) – address prefix name to create

• vpc (str) – VPC name

• zone (str) – zone name

• cidr (str) – CIDR for address prefix

ocs_ci.utility.ibmcloud.create_cluster(cluster_name)

Create OCP cluster.

Parameters:

cluster_name (str) – Cluster name.

Raises:

UnexpectedBehaviour – in the case, the cluster is not installed successfully.

ocs_ci.utility.ibmcloud.create_public_gateway(gateway_name, vpc, zone, resource_group)

Create public gateway in VPC.

Parameters:

• gateway_name (str) – public gateway name

• vpc (str) – VPC name

• zone (str) – zone name

• resource_group (str) – resource group name

ocs_ci.utility.ibmcloud.create_resource_group(resource_group)

Create resource group.

Parameters:

resource_group (str) – resource group name

ocs_ci.utility.ibmcloud.create_subnet(subnet_name, vpc, zone, cidr, resource_group)

Create subnet in VPC.

Parameters:

• subnet_name (str) – address prefix name to create

• vpc (str) – VPC name

• zone (str) – zone name

• cidr (str) – CIDR for address prefix

• resource_group (str) – resource group name

ocs_ci.utility.ibmcloud.create_vpc(cluster_name, resource_group)

Create VPC.

Parameters:

• cluster_name (str) – cluster name

• resource_group (str) – resource group name

ocs_ci.utility.ibmcloud.delete_account_policy(policy_id, token=None)

Delete account policy

Parameters:

• policy_id (str) – policy ID

• token (str) – IBM Cloud token to be used for API calls - if not provided it will create new one.

Returns:

True in case it successfully deleted

Return type:

bool

Raises:

APIRequestError – in case API call didn’t went well

ocs_ci.utility.ibmcloud.delete_dns_records(cluster_name)

Delete DNS records leftover from cluster destroy.

Parameters:

cluster_name (str) – Name of the cluster, used to filter DNS records

ocs_ci.utility.ibmcloud.delete_service_id(service_id)

Delete service ID

Parameters:

service_id (str) – ID of service ID to delete

ocs_ci.utility.ibmcloud.destroy_cluster(cluster)

Destroy the cluster on IBM Cloud.

Parameters:

cluster (str) – Cluster name or ID.

ocs_ci.utility.ibmcloud.find_free_network_subnets(subnet_cidr, network_prefix=27)

This function will look for currently used subnet, and will try to find one which is not occupied by any other VPC.

Parameters:

• subnet_cidr (str) – subnet CIDR in which range to look for free subnet (e.g. 10.240.0.0/18)

• network_prefix (int) – subnet prefix to look for

Returns:

(network_with_prefix, network_split1, network_split2), where

network_with_prefix - is network CIDR which we are looking for. network_split1 - is first CIDR split of network_with_prefix network_split2 - is second CIDR split of network_with_prefix

Return type:

tuple

ocs_ci.utility.ibmcloud.get_api_token()

Get IBM Cloud API Token for API Calls authentication

Returns:

IBM Cloud API Token

Return type:

str

ocs_ci.utility.ibmcloud.get_bucket_regions_map()

Fetches the buckets and their regions

Returns:

Dictionary with bucket name as Key and region as value

Return type:

dict

ocs_ci.utility.ibmcloud.get_cluster_account_policies(cluster_name, cluster_service_ids)

Get cluster account policies.

Parameters:

• cluster_name (str) – cluster name

• cluster_service_ids (list) – list of service IDs, e.g. output from get_cluster_service_ids.

Returns:

Account policies for cluster

Return type:

list

ocs_ci.utility.ibmcloud.get_cluster_details(cluster)

Returns info about the cluster which is taken from the ibmcloud command.

Parameters:

cluster (str) – Cluster name or ID

ocs_ci.utility.ibmcloud.get_cluster_service_ids(cluster_name, get_infra_id_from_metadata=True)

Get cluster service IDs

Parameters:

• cluster_name (str) – cluster name

• get_infra_id_from_metadata (bool) – if set to true it will try to get infra ID from metadata.json file (Default: True)

Returns:

service IDs for cluster

Return type:

list

ocs_ci.utility.ibmcloud.get_ibmcloud_cluster_region()

Get IBM Cloud region from the cluster’s infrastructure object.

This function queries the live cluster to retrieve the IBM Cloud region from the infrastructure status, which may differ from the metadata.json file.

Returns:

IBM Cloud region where the cluster is deployed

Return type:

str

Raises:

CommandFailed – If the oc command fails

ocs_ci.utility.ibmcloud.get_ibmcloud_ocp_version()

Get OCP version available in IBM Cloud.

ocs_ci.utility.ibmcloud.get_kubeconfig(cluster, path)

Export kubeconfig to provided path.

Parameters:

• cluster (str) – Cluster name or ID.

• path (str) – Path where to create kubeconfig file.

ocs_ci.utility.ibmcloud.get_region(cluster_path)

Get region from metadata.json in given cluster_path

Parameters:

cluster_path – path to cluster install directory

Returns:

region where cluster is deployed

Return type:

str

ocs_ci.utility.ibmcloud.get_resource_group_name(cluster_path)

Get resource group from metadata.json in given cluster_path

Parameters:

cluster_path – path to cluster install directory

Returns:

resource group name

Return type:

str

ocs_ci.utility.ibmcloud.get_security_group_id(sg_name, vpc_id='', resource_group_id='', resource_group_name='')

Get security group ID by name

Parameters:

• sg_name (str) – security group name

• vpc_id (str) – VPC ID to filter security groups, if empty it will return security groups for all VPCs.

• resource_group_id (str) – Resource group ID to filter security groups.

• resource_group_name (str) – Resource group name to filter security groups

Returns:

security group ID or empty string if not found

Return type:

str

ocs_ci.utility.ibmcloud.get_security_group_name_by_pattern(sg_name_pattern, vpc_id='', resource_group_id='', resource_group_name='')

Get security group name by pattern

Parameters:

• sg_name_pattern (str) – security group name pattern (regular expression)

• vpc_id (str) – VPC ID to filter security groups, if empty it will return security groups for all VPCs

• resource_group_id (str) – Resource group ID to filter security groups

• resource_group_name (str) – Resource group name to filter security groups

Returns:

security group name or empty string if not found

Return type:

str

ocs_ci.utility.ibmcloud.get_security_groups(vpc_id='', resource_group_id='', resource_group_name='')

Get security groups in IBM Cloud

Parameters:

• vpc_id (str) – VPC ID to filter security groups, if empty it will return security groups for all VPCs.

• resource_group_id (str) – Resource group ID to filter security groups.

• resource_group_name (str) – Resource group name to filter security groups.

Returns:

security groups

Return type:

list

ocs_ci.utility.ibmcloud.get_used_subnets(vpc_id='')

Get currently used subnets in IBM Cloud

Parameters:

vpc_id (str) – VPC ID to filter subnets. Empty string means all VPCs.

Returns:

subnets

Return type:

list

ocs_ci.utility.ibmcloud.get_worker_floating_ips()

Retrieve a mapping of worker node names to their associated floating IPs.

Returns:

A dictionary mapping worker node names to their floating IP addresses.

Return type:

dict

ocs_ci.utility.ibmcloud.is_cluster_installed(cluster)

Check if cluster is installed and return True if so, False otherwise.

Parameters:

cluster (str) – Cluster name or ID

ocs_ci.utility.ibmcloud.is_ibm_platform()

Check if cluster is IBM or Not

ocs_ci.utility.ibmcloud.label_nodes_region()

Apply the region label to the worker nodes. Necessary for IBM COS-backed backingstore support.

ocs_ci.utility.ibmcloud.list_clusters(provider=None)

Returns info about the cluster which is taken from the ibmcloud command.

Parameters:

provider (str) – Provider type (classic, vpc-classic, vpc-gen2).

ocs_ci.utility.ibmcloud.login(region=None, resource_group=None)

Login to IBM Cloud cluster

Parameters:

• region (str) – region to log in, if not specified it will use one from config

• resource_group (str) – resource group to log in, if not specified it will use one from config or nothing if not defined

ocs_ci.utility.ibmcloud.open_ports_on_ibmcloud_hub_cluster()

Add the inbound rules for these cluster security configs -cluster-wide and -openshift-net to open the following ports: 3300, 6789, 9283, 6800-7300 and 31659

ocs_ci.utility.ibmcloud.remove_nfs_lb_security_group_rules()

Remove inbound TCP 2049 rules from the security groups attached to the NFS LoadBalancer on IBM Cloud VPC.

Should be called before deleting the NFS LoadBalancer Service so the VPC LB is still present for look-up.

ocs_ci.utility.ibmcloud.run_ibmcloud_cmd(cmd, secrets=None, timeout=600, ignore_error=False, **kwargs)

Wrapper function for run_cmd which if needed will perform IBM Cloud login command before running the ibmcloud command. In the case run_cmd will fail because the IBM cloud got disconnected, it will login and re-try.

Parameters:

• cmd (str) – command to run

• secrets (list) – A list of secrets to be masked with asterisks This kwarg is popped in order to not interfere with subprocess.run(**kwargs)

• timeout (int) – Timeout for the command, defaults to 600 seconds.

• ignore_error (bool) – True if ignore non zero return code and do not raise the exception.

ocs_ci.utility.ibmcloud.set_region(region=None)

Sets the cluster region to ENV_DATA when enable_region_dynamic_switching is enabled.

Parameters:

region (str) – region to set, if not defined it will try to get from metadata.json

ocs_ci.utility.ibmcloud.set_resource_group_name(resource_group_name=None)

Sets the resource group to ibmcloud cli

Parameters:

resource_group_name (str) – Resource Group Name

ocs_ci.utility.ibmcloud.set_target_region()

Set the target region for the IBM Cloud CLI.

ocs_ci.utility.ibmcloud_bm module

Module for interactions with IBM Cloud Cluster.

class ocs_ci.utility.ibmcloud_bm.IBMCloudBM(region=None)

Bases: object

Wrapper for IBM Cloud with Bare metal machines

get_all_machines()

Get all the machines in the IBMCloud Bare metal machines

Returns:

List of dictionaries. List of all the machines in the IBMCloud Bare metal machines

Return type:

list

get_machines_by_names(machine_names)

Get the machines in the IBMCloud Bare metal machines that have the given machine names

Parameters:

machine_names (list) – The list of the machine names to search for.

Returns:

Get the machines in the IBMCloud Bare metal machines that have the given machine names

get_machines_that_are_not_off_from_sensor_data(machines)

Retrieve a server’s hardware state via its internal sensors and check if RPM Sensor data is present. Data from RPM sensors won’t be available if the machine is OFF. To be used only as an alternate method. A command resulting in the power status of the machine is not available. This method is not recommended to identify the machines that are ON and functional

Parameters:

machines (list) – List of the IBMCLoud Bare metal machines objects

Returns:

List of machines which are not off

Return type:

list

login()

Login to IBM Cloud account

restart_machines(machines, force=False)

Reboot the IBMCloud Bare metal machines

Parameters:

• machines (list) – List of the IBMCLoud Bare metal machines objects to restart

• force (bool) – If False, will perform a soft reboot. Otherwise, if True, will perform a hard reboot

restart_machines_by_stop_and_start(machines)

Restart the IBMCloud Bare metal machines by stop and start

Parameters:

machines (list) – List of the IBMCLoud Bare metal machines objects to restart

run_ibmcloud_bm_cmd(cmd, secrets=None, timeout=600, ignore_error=False, **kwargs)

Wrapper function for run_cmd which if needed will perform IBM Cloud login command before running the ibmcloud bare metal command. In the case run_cmd will fail because the IBM cloud got disconnected, it will login and re-try.

Parameters:

• cmd (str) – command to run

• secrets (list) – A list of secrets to be masked with asterisks This kwarg is popped in order to not interfere with subprocess.run(**kwargs)

• timeout (int) – Timeout for the command, defaults to 600 seconds.

• ignore_error (bool) – True if ignore non zero return code and do not raise the exception.

start_machines(machines)

Start the IBMCloud Bare metal machines

Parameters:

machines (list) – List of the IBMCLoud Bare metal machines objects to start

stop_machines(machines)

Stop the IBMCloud Bare metal machines

Parameters:

machines (list) – List of the IBMCLoud Bare metal machines objects to stop

ocs_ci.utility.ipam module

This module will interact with IPAM server

class ocs_ci.utility.ipam.IPAM(appiapp, ipam=None, token=None)

Bases: object

IPAM class

assign_ip(host, subnet)

Reserve IP in IPAM server for a given host

Parameters:

• host (str) – hostname to reserve IP in IPAM server

• subnet (str) – subnet to reserve IP

Returns:

Reserved IP

Return type:

str

Raises:

• HTTPError – in case of HTTP error

• ConnectionError – in case of Connection Error

• Timeout – in case of Timeout

• RequestException – Any Exception from requests

• IPAMAssignUpdateFailed – if it fails to assign IP

assign_ips(hosts, subnet)

Assign IPs to the hosts

Parameters:

• hosts (list) – List of hosts to reserve IP in IPAM server

• subnet (str) – subnet to reserve IPs

Returns:

List of Reserved IP’s

Return type:

list

release_ip(hostname)

Release IP from IPAM server

Parameters:

hostname (str) – Hostname to release IP

Raises:

IPAMReleaseUpdateFailed – If it fails to release IP from IPAM server

release_ips(hosts)

Releases host IP’s from IPAM server

Parameters:

hosts (list) – List of host names to release IP’s

ocs_ci.utility.iscsi_config module

This module is used for configuring iscsi.

ocs_ci.utility.iscsi_config.cleanup_iscsi_target(target_node_ssh, target_iqn, worker_iqns, wipe_data, username_target)

Complete cleanup of iSCSI target.

Parameters:

• target_node_ssh (object) – An established SSH connection to the target VM.

• target_iqn (str) – Target IQN

• worker_iqns (list) – List of worker IQNs to remove

• wipe_data (bool) – Whether to wipe data (default: True)

• username_target (str) – SSH user for target VM

ocs_ci.utility.iscsi_config.configure_initiators(worker_node_names)

Configures a worker node as an iSCSI initiator. This function installs necessary packages, discovers targets, logs in to the target, and enables the iSCSI service.

Parameters:

worker_node_names (list) – List of worker node names (strings).

Raises:

NodeNotFoundError – If the node IP cannot be retrieved (none or empty).

ocs_ci.utility.iscsi_config.configure_target(target_node_ssh, target_iqn, worker_iqns)

Configures the iSCSI target with given IQNs and IP.

Parameters:

• target_node_ssh (object) – An established SSH connection to the target VM.

• target_iqn (str) – Target iSCSI IQN.

• worker_iqns (list) – List of IQNs of worker nodes.

ocs_ci.utility.iscsi_config.get_worker_iqns(worker_node_names)

Collects iSCSI initiator IQNs from worker nodes. This function ensures the iSCSI service is started before reading the IQN.

Parameters:

worker_node_names (list) – List of worker node names (strings).

Returns:

A list of IQNs of worker nodes or an empty list on failure.

Return type:

list

ocs_ci.utility.iscsi_config.get_worker_node_names()

Get worker node names using ocs-ci node utilities. This function works on pure OCP clusters without OCS installed.

Returns:

A list of worker node names (strings).

Return type:

list

ocs_ci.utility.iscsi_config.iscsi_setup()

Setup iSCSI target and initiators.

ocs_ci.utility.iscsi_config.iscsi_teardown()

Tear down iSCSI target.

ocs_ci.utility.iscsi_config.remove_acls_from_target(target_node_ssh, target_iqn, worker_iqns, username)

Remove ACLs and LUN mappings from target using targetcli.

Parameters:

• target_node_ssh (object) – An established SSH connection to the target VM.

• target_iqn (str) – Target iSCSI IQN.

• worker_iqns (list) – List of IQNs of worker nodes.

• username (str) – Username for the target VM.

ocs_ci.utility.iscsi_config.run_iscsi_setup_validations()

Run all three iSCSI validations: connectivity, sessions, and devices. Called at the end of iscsi_setup() and can be used by tests for post-deploy verification.

Raises:

• AssertionError – If any validation fails.

• ValueError – If iSCSI not configured in ENV_DATA.

• ValueError – If no worker nodes are found.

ocs_ci.utility.iscsi_config.validate_iscsi_connectivity()

Validate network connectivity to iSCSI target from all worker nodes.

Raises:

• AssertionError – If any worker node cannot connect.

• ValueError – If iSCSI target IP is not configured in ENV_DATA.

• ValueError – If no worker nodes are found.

ocs_ci.utility.iscsi_config.validate_iscsi_devices()

Validate iSCSI devices are visible on worker nodes.

Raises:

• AssertionError – If no iSCSI devices are found on any node.

• ValueError – If iSCSI target IP or IQN is not configured in ENV_DATA.

• ValueError – If no worker nodes are found.

ocs_ci.utility.iscsi_config.validate_iscsi_sessions()

Validate iSCSI sessions are established on all worker nodes.

Raises:

AssertionError – If any worker node does not have an iSCSI session.

ocs_ci.utility.iscsi_config.verify_cleanup(target_node_ssh, target_iqn, username_target, worker_iqns)

Verify cleanup was successful.

Parameters:

• target_node_ssh (object) – An established SSH connection to the target VM.

• target_iqn (str) – Target iSCSI IQN.

• username_target (str) – Username for the target VM.

• worker_iqns (list) – List of IQNs of worker nodes.

Raises:

UnexpectedBehaviour – If an ACL still exists after cleanup (verification failed).

ocs_ci.utility.iscsi_config.verify_iscsi_devices(worker_node_names, target_iqn)

Verify iSCSI devices are visible on worker nodes.

This function detects iSCSI block devices by: 1. Using lsblk to find devices with transport type ‘iscsi’

Parameters:

• worker_node_names (list) – List of worker node names (strings).

• target_iqn (str) – Target iSCSI IQN to verify.

Returns:

Dictionary with node name as key and device information as value.

Return type:

dict

ocs_ci.utility.iscsi_config.verify_iscsi_sessions(worker_node_names, target_iqn)

Verify iSCSI sessions are established on worker nodes.

Parameters:

• worker_node_names (list) – List of worker node names (strings).

• target_iqn (str) – Target iSCSI IQN to verify.

Returns:

Dictionary with node name as key and verification result as value.

Return type:

dict

ocs_ci.utility.iscsi_config.verify_iscsi_setup()

Comprehensive verification of iSCSI setup.

Returns:

Dictionary containing all verification results.

Return type:

dict

ocs_ci.utility.iscsi_config.verify_iscsi_target_connectivity(worker_node_names, target_ip, target_port=3260)

Verify network connectivity to iSCSI target from worker nodes.

Parameters:

• worker_node_names (list) – List of worker node names (strings).

• target_ip (str) – iSCSI target IP address.

• target_port (int) – iSCSI target port (default: 3260).

Returns:

Dictionary with node name as key and connectivity result as value.

Return type:

dict

ocs_ci.utility.iscsi_config.wipe_luns_on_target(target_node_ssh, target_iqn, username)

Wipe LUN data on target VM. Also deletes IBM Spectrum Scale related resources. Only wipes devices that are referenced in LocalDisks resources.

Parameters:

• target_node_ssh (object) – An established SSH connection to the target VM.

• target_iqn (str) – Target iSCSI IQN.

• username (str) – Username for the target VM.

Raises:

• Exception – If there is an error while deleting StorageClasses or LocalDisks.

• Exception – If there is an error while wiping the devices.

• Exception – If there is an error while removing signatures from the devices.

• Exception – If there is an error while verifying the cleanup.

• Exception – If there is an error while deleting the StorageClasses or LocalDisks.

ocs_ci.utility.jira module

class ocs_ci.utility.jira.JiraHelper

Bases: object

Simple Jira integration for OCS-CI. Requires a dict with keys: url, username, password.

add_comment(issue_key: str, text: str)

Add a comment to an issue.

Parameters:

• issue_key (str) – The key of the Jira issue e.g. ‘DFBUGS-2781’

• text (str) – The text of the comment

Returns:

A dictionary containing the complete JSON of the Jira issue

Return type:

dict

get_issue(issue_key: str) → dict

Return complete JSON of the Jira issue.

Parameters:

issue_key (str) – The key of the Jira issue e.g. ‘DFBUGS-2781’

Returns:

A dictionary containing the complete JSON of the Jira issue

Return type:

dict

ocs_ci.utility.json module

class ocs_ci.utility.json.SetToListJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: JSONEncoder

The CustomJSONEncoder class is a subclass of json.JSONEncoder designed to handle the serialization of Python objects into JSON format, with a specific focus on converting set objects into lists. This is necessary because the default JSON encoder in Python does not support set objects, which are not a valid JSON data type. This way we avoid “TypeError: Object of type set is not JSON serializable”

Usage:

json.dumps(data, cls=SetToListJSONEncoder)

default(obj)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return super().default(o)

ocs_ci.utility.kms module

This module contains KMS related class and methods currently supported KMSs: Vault and HPCS

class ocs_ci.utility.kms.AzureKV(namespace='openshift-storage')

Bases: KMS

Represents an Azure Key Vault implementation of KMS.

azure_kv_osd_keys()

List of OSD keys found in Azure Key Vault

azure_kv_secrets()

List the secrets in the Azure Key Vault.

create_azure_kv_csi_kms_connection_details()

Create Azure specific csi-kms-connection-details configmap resource

create_azure_kv_ocs_csi_kms_connection_details()

Creates Azure KV OCS CSI KMS connection details ConfigMap.

create_azure_kv_secrets(prefix='azure-ocs-')

Creates Azure KV secrets.

deploy()

This Function will create the Azure KV connection details in the ConfigMap.

is_azure_kv_connection_exists()

Checks if the Azure KV connection exists in the ConfigMap

post_deploy_verification()

Post Deploy Verification For Azure Key Vault.

remove_kmsid()

Removing azure kmsid from the configmap csi-kms-connection-details.

Returns:

True if KMS ID is successfully removed, otherwise False.

Return type:

bool

verify_osd_keys_present_on_azure_kv()

Verify if all OSD keys are present in Azure Key Vault

verify_pv_secrets_present_in_azure_kv(vol_handle)

Verify Azure KV has the secrets for given volume handle.

Returns:

True if PV secrets are found in the Azure KV, otherwise False.

Return type:

bool

class ocs_ci.utility.kms.HPCS

Bases: KMS

A class which handles deployment and other configs related to HPCS

cleanup()

Cleanup the backend resources in case of external

create_hpcs_csi_kms_connection_details(namespace='openshift-storage')

Create hpcs specific csi kms connection details configmap resource

create_ibm_kp_kms_secret(namespace='openshift-storage')

create hpcs specific csi kms secret resource

create_ocs_hpcs_resources()

This function takes care of creating ocp resources for secrets like hpcs customer root key, service api key, etc. Assumption is hpcs section in AUTH file contains hpcs service instance id, base url, token url, api key and customer root key.

delete_resource(resource_name, resource_type, resource_namespace)

Given resource type, resource name and namespace, this function will delete oc resource

Parameters:

• resource_name – name of the resource

• resource_type – type of resource such as secret

• resource_namespace – namespace in which resource is present

deploy()

This function delegates the deployment of hpcs based on OCP or vault standalone external mode deployment

deploy_hpcs_external()

This function takes care of deployment and configuration for external mode hpcs deployment. We are assuming that an external hpcs service already exists and we will be just configuring the necessary OCP objects for OCS like secrets, token etc

gather_hpcs_config()

This function populates the hpcs configuration

gather_init_hpcs_conf()

Gather hpcs configuration and init the vars This function currently gathers only for external mode

get_token_for_ibm_api_key()

This function retrieves the access token in exchange of an IBM API key

Returns:

access token for authentication with IBM endpoints

Return type:

(str)

list_hpcs_keys()

This function lists the keys present in a HPCS instance

Returns:

list of keys in a HPCS instance

Return type:

(list)

post_deploy_verification()

Validating the OCS deployment from hpcs perspective

validate_external_hpcs()

This function is for post OCS deployment HPCS verification

Following checks will be done 1. check osd encryption keys in the HPCS path 2. check noobaa keys in the HPCS path 3. check storagecluster CR for ‘kms’ enabled

Raises:

NotFoundError – if key not found in HPCS OR in the resource CR

class ocs_ci.utility.kms.KMIP

Bases: KMS

A class which handles deployment and other configs related to KMIP (Thales CipherTrust Manager)

check_key_exists_in_ciphertrust(key_id)

Check if a key with the key_id is present in CipherTrust Manager

Parameters:

key_id (str) – ID of the key to be checked for

Returns:

True, if the key exists in CipherTrust Manager

Return type:

(bool)

cleanup()

Cleanup for KMIP

create_ciphertrust_key(key_name)

Create a key in Ciphertrust Manager to be used for PV encryption

Parameters:

key_name (str) – Name of the key to be created

Returns:

ID of the key created in CipherTrust

Return type:

(str)

create_kmip_csi_kms_connection_details(namespace='openshift-storage')

Create KMIP specific csi-kms-connection-details configmap resource

create_kmip_secret(type='ocs')

Create secret containing the certificates and unique identifier (only for PV encryption) from CipherTrust Manager KMS

Parameters:

type (str) – csi, if the secret is being created for PV/Storageclass encryption ocs, if the secret is for clusterwide encryption

Returns:

name of the kmip secret

Return type:

(str)

create_odf_kmip_resources()

Create secret containing certs and the ocs-kms-connection-details confignmap

delete_ciphertrust_key(key_id)

Delete key from CipherTrust Manager

Parameters:

key_id (str) – ID of the key to be deleted

deploy()

This function delegates the deployment of KMS using KMIP. Thales CipherTrust Manager is the only supported vendor for now.

deploy_kmip_ciphertrust()

This function configures the resources required to use Thales CipherTrust Manager with ODF

get_key_info_ciphertrust(key_id)

Retrieve information about a given key

Parameters:

key_id (str) – ID of the key in CipherTrust

Returns

(dict): Dictionary with key details

get_key_list_ciphertrust(limit=100)

Lists all keys in CipherTrust Manager

Parameters:

limit (int) – number of entries to limit the results

Returns:

list containing the IDs of the keys

Return type:

(list)

get_noobaa_key_id()

Retrieve the key ID used for encryption by Noobaa

Returns:

Key ID used by NooBaa for encryption

Return type:

(str)

get_osd_key_ids()

Retrieve the key ID used for OSD encryption stored in their respective secrets

Returns:

list of key IDs from all OSDs

Return type:

(list)

post_deploy_verification()

Verify ODF deployment using KMIP

update_kmip_env_vars()

Set the environment variable for CipherTrust to allow running ksctl CLI cmds

validate_ciphertrust_deployment()

Verify whether OSD and NooBaa keys are stored in CipherTrust Manager

class ocs_ci.utility.kms.KMS(provider=None)

Bases: object

This is base class for any KMS integration

create_csi_kms_resources()

create_resource(resource_data, prefix=None)

Given a dictionary of resource data, this function will creates oc resource

Parameters:

• resource_data (dict) – yaml dictionary for resource

• prefix (str) – prefix for NamedTemporaryFile

deploy()

post_deploy_verification()

class ocs_ci.utility.kms.Vault

Bases: KMS

A class which handles deployment and other configs related to vault

HCP is enterprise vault, clusters are deployed on AWS platform can communicate with no error. for vsphere we use our community vault

cleanup()

Cleanup the backend resources in case of external

create_namespace(vault_namespace)

Create a vault namespace

Parameters:

vault_namespace (str) – name of the vault namespace

Raises:

VaultOperationError – If namespace is not created successfully

create_ocs_kube_auth_resources(sa_name='odf-vault-auth')

This function will create the serviceaccount and clusterrolebindings required for kubernetes auth

Parameters:

sa_name (str) – Name of the service account in ODF

create_ocs_vault_cert_resources()

Explicitly create secrets like ca cert, client cert, client key Assumption is vault section in AUTH file contains base64 encoded (with padding) ca, client certs, client key

create_ocs_vault_resources()

This function takes care of creating ocp resources for secrets like ca cert, client cert, client key and vault token Assumption is vault section in AUTH file contains base64 encoded (with padding) ca, client certs, client key and vault path token

create_tenant_configmap(tenant_namespace, **vault_config)

This functional will create a configmap in the tenant namespace to override the vault config in csi-kms-connection-details configmap.

Parameters:

• tenant_namespace (str) – Tenant namespace

• vaultBackend (str) – KV version to be used, either kv or kv-v2

• vaultBackendPath (str) – The backend path in Vault where the encryption keys will be stored

• vaultNamespace (str) – Namespace in Vault, if exists, where the backend path is created

• vaultRole (str) – (Vaulttenantsa) The role name in Vault configured with kube auth method for the given policy and tenant namespace

• vaultAuthPath (str) – (Vaulttenantsa) The path where kubernetes auth method is enabled

• vaultAuthNamespace (str) – (Vaulttenantsa) The namespace where kubernetes auth method is enabled, if exists

create_tenant_sa(namespace)

This function will create the serviceaccount in the tenant namespace to authenticate to Vault when vaulttenantsa KMS type is used for PV encryption.

Parameters:

namespace (str) – The tenant namespace where the service account will be created

create_token_reviewer_resources()

This function will create the rbd-csi-vault-token-review SA, clusterRole and clusterRoleBindings required for the kubernetes auth method with vaulttenantsa encryption type.

Raises:

CommandFailed – Exception if the command fails

create_vault_csi_kms_connection_details(kv_version, vault_auth_method='vaulttokens', namespace=None)

Create vault specific csi kms connection details configmap resource

create_vault_csi_kms_token(namespace=None)

create vault specific csi kms secret resource

Parameters:

namespace (str) – then value from config will be used.

create_vault_kube_auth_role(namespace, role_name='csi-kubernetes', sa_name='ceph-csi-vault-sa')

Create a role for tenant authentication in Vault

Parameters:

• namespace (str) – namespace in ODF cluster

• role_name (str) – Name of the role in Vault

• sa_name (str) – Service account in the tenant namespace to be used for authentication

deploy()

This function delegates the deployment of vault based on OCP or vault standalone external mode deployment

deploy_vault_external()

This function takes care of deployment and configuration for external mode vault deployment. We are assuming that an external vault service already exists and we will be just configuring the necessary OCP objects for OCS like secrets, token etc

deploy_vault_internal()

This function takes care of deployment and configuration for internal mode vault deployment on OCP

gather_init_vault_conf()

Gather vault configuration and init the vars This function currently gathers only for external mode Gathering for internal mode woulde be different

gather_vault_config()

This function populates the vault configuration

generate_vault_token(ttl='768h')

Generate a token for self.vault_policy_name

Returns:

vault token

Return type:

str

get_noobaa_secret()

Fetches the NooBaa backend secret from the Vault.

Returns:

The NooBaa backend secret.

Return type:

str

get_osd_secret(device_handle)

Fetch the OSD encryption key for the given device handle from Vault.

Parameters:

device_handle (str) – The device handle for which to retrieve the OSD secret.

Returns:

The OSD encryption secret if found, otherwise None.

Return type:

str

get_pv_secret(device_handle, kms_id=None)

Get secret stored in the vault KMS for the given device_handle

Parameters:

• device_handle (str) – PV device handle string

• kms_id (str, optional) – The encryption KMS ID from PV volumeAttributes. If provided, uses the backend path specific to this KMS ID. If not provided, falls back to the default backend path.

Returns:

passphrase stored in the vault KMS for given device handle.

Return type:

secret (str)

get_vault_backend_path(resource_configmap='ocs-kms-connection-details')

Fetch the vault backend path used for this deployment This can be obtained from kubernetes secret resource ‘ocs-kms-connection-details’

apiVersion: v1
data:
  KMS_PROVIDER: vault
  KMS_SERVICE_NAME: vault
  VAULT_ADDR: https://xx.xx.xx.xx:8200
  VAULT_BACKEND_PATH: ocs

get_vault_backend_path_for_kms_id(kms_id)

Get the Vault backend path for a specific KMS ID from csi-kms-connection-details.

Parameters:

kms_id (str) – The encryption KMS ID from PV volumeAttributes

Returns:

The VAULT_BACKEND_PATH for the given KMS ID, or None if not found

Return type:

str

get_vault_connection_info(resource_name=None, resource_configmap='ocs-kms-connection-details')

Get resource info from ocs-kms-connection-details or csi-kms-connection-details ConfigMap.

Parameters:

resource_name (str) – name of the resource

Returns:

The resource information, or None if not found.

Return type:

str or None

get_vault_kube_auth_role()

Fetch the role name from ocs-kms-connection-details configmap

get_vault_namespace()

From kms connection details resource obtain namespace

get_vault_path_token()

Fetch token from kubernetes secret we need this to find the vault policy default name in case of ocs is ‘ocs-kms-token’

apiVersion: v1
data:
  token: cy5DRXBKV0lVbzNFQjM1VHlGMFNURzZQWms=
kind: Secret
metadata:
  name: ocs-kms-token
namespace: openshift-storage
type: Opaque

property kms_vault_policy_name

post_deploy_verification()

Validating the OCS deployment from vault perspective

remove_vault_backend_path(vault_namespace=None)

remove vault path

Parameters:

vault_namespace (str) – Namespace in Vault, if exists, where the backend path is created

remove_vault_namespace()

Cleanup the namespace

Raises:

KMSResourceCleanupError – If namespace deletion fails

remove_vault_policy(vault_namespace=None)

Cleanup the policy we used

Parameters:

namespace (vault) – Namespace in Vault, if exists, where the backend path is created

set_vault_policy()

Set the policy name matching the cluster name or one provided in config

setup_vault_client_cert()

For Vault cli interaction with the server we need client cert to talk to HTTPS on the vault server

update_vault_env_vars()

In order to run vault CLI we need following env vars VAULT_ADDR and VAULT_TOKEN

validate_external_vault()

This function is for post OCS deployment vault verification

Following checks will be done 1. check osd encryption keys in the vault path 2. check noobaa keys in the vault path 3. check storagecluster CR for ‘kms’ enabled

Raises:

NotFoundError – if key not found in vault OR in the resource CR

vault_backend_path_exists(backend_path)

Check if vault backend path already exists

Parameters:

backend_path (str) – name of the vault backend path

Returns:

True if exists else False

Return type:

bool

vault_create_backend_path(backend_path=None, kv_version=None)

create vault path to be used by OCS

Raises:

VaultOperationError exception –

vault_create_namespace(namespace=None)

Create a vault namespace if it doesn’t exists

vault_create_policy(policy_name=None)

Create a vault policy and generate token

Raises:

VaultOperationError exception –

vault_kube_auth_setup(auth_path=None, auth_namespace=None, token_reviewer_name='rbd-csi-vault-token-review')

Setup kubernetes auth method in Vault

Parameters:

• auth_path (str) – The path where kubernetes auth is to be enabled. If not provided default ‘kubernetes’ path is used

• auth_namespace (str) – The vault namespace where kubernetes auth is to be enabled, if applicable

• token_reviewer_name (str) – Name of the token-reviewer serviceaccount in openshift-storage namespace

Raises:

VaultOperationError – if kube auth method setup fails

vault_namespace_exists(vault_namespace)

Check if vault namespace already exists

Parameters:

vault_namespace (str) – name of the vault namespace

Returns:

True if exists else False

Return type:

bool

property vault_path_token

vault_sealed()

Returns:

if vault is sealed then return True else False

Return type:

bool

vault_unseal()

Unseal vault if sealed

Raises:

VaultOperationError – In case unseal operation failed

ocs_ci.utility.kms.fetch_noobaa_secret_from_vault(vault_backend_path)

Fetches the NooBaa backend secret from the Vault.

Parameters:

vault_backend_path (str) – Vault Backend Path

Returns:

The NooBaa backend secret.

Return type:

str

Raises:

UnexpectedBehaviour – If the NooBaa secret key is not found in the Vault response.

ocs_ci.utility.kms.fetch_osd_secret_from_vault(device_handle, vault_backend_path)

Fetch the OSD encryption key for the given device handle from Vault.

Parameters:

• device_handle (str) – The device handle for which to retrieve the OSD secret.

• vault_backend_path (str) – Vault Backend Path.

Returns:

The OSD encryption secret if found, otherwise None.

Return type:

str

Raises:

UnexpectedBehaviour – If the secret key is not found in the Vault response.

ocs_ci.utility.kms.get_encryption_kmsid()

Get encryption kmsid from ‘csi-kms-connection-details’ configmap resource

Returns:

A list of KMS IDs available

Return type:

kmsid (list)

Raises:

KMSConnectionDetailsError – if csi kms connection detail doesn’t exist

ocs_ci.utility.kms.get_kms_deployment()

ocs_ci.utility.kms.get_kms_details()

ocs_ci.utility.kms.get_kms_endpoint()

Fetch VAULT_ADDR or KMIP_ENDPOINT if kmip provider from ocs-kms-connection-details configmap.

Returns:

KMS endpoint address

Return type:

str

ocs_ci.utility.kms.get_ksctl_cli(bin_dir=None)

Download ksctl to interact with CipherTrust Manager via CLI

Parameters:

bin_dir (str) – Path to bin directory (default: config.RUN[‘bin_dir’])

ocs_ci.utility.kms.is_key_present_in_path(key, path)

Check if key is present in the backend Path

Parameters:

• key (str) – Name of the key

• path (str) – Vault backend path name

Returns:

True if key is present in the backend path

Return type:

(bool)

ocs_ci.utility.kms.is_kms_enabled(dont_raise=False)

Checks StorageCluster yaml if kms is configured.

Returns:

True if KMS is configured else False

Return type:

(bool)

ocs_ci.utility.kms.remove_kmsid(kmsid)

This function will remove all the details for the given kmsid from the csi-kms-connection-details configmap

Parameters:

kmsid (str) – kmsid to be remove_kmsid

Raises:

KMSResourceCleaneupError – If the kmsid entry is not deleted

ocs_ci.utility.kms.remove_token_reviewer_resources()

Delete the SA, clusterRole and clusterRoleBindings for token reviewer

ocs_ci.utility.kms.set_kms_endpoint(address)

Set VAULT_ADDR or KMIP_ENDPOINT if kmip provider in ocs-kms-connection-details configmap to provided value

Parameters:

address (str) – Address to be set in KMS configuration

ocs_ci.utility.kms.update_csi_kms_vault_connection_details(update_config)

Update the vault connection details in the resource csi-kms-connection-details

Parameters:

update_config (dict) – A dictionary of vault info to be updated

ocs_ci.utility.kms.vault_kv_list(path)

List kv from a given path

Parameters:

path (str) – Vault backend path name

Returns:

of kv present in the path

Return type:

list

ocs_ci.utility.kubevirt_vm module

Module for interactions with Kubevirt hosted cluster.

class ocs_ci.utility.kubevirt_vm.KubevirtVM(cluster_name)

Bases: object

Wrapper for VM objects with kubevirt. The class should be used in Provider mode when we have provider and client clusters in the run.

get_all_kubevirt_vms()

Get all the kubevirt VMs in the cluster

Returns:

List of dictionaries. List of all the VM objects in the cluster.

Return type:

list

get_kubevirt_vms_by_names(vm_names)

Get the VMs that have the given VM names

Parameters:

vm_names (list) – The list of the VM names to search for.

Returns:

Get the VMs that have the given VM names

Return type:

list

pause_kubevirt_vms(vms, wait=True)

Pause the kubevirt VMs

Parameters:

• vms (list) – List of the kubevirt VM objects to pause

• wait (bool) – If true, wait for VM to pause. False, otherwise.

restart_kubevirt_vms(vms, wait=True)

Restart the kubevirt VMs

Parameters:

• vms (list) – List of the kubevirt VM objects to restart

• wait (bool) – If true, wait for VM to start. False, otherwise.

restart_kubevirt_vms_by_stop_and_start(vms, wait=True, force=False)

Restart the VMs by stop and start

Parameters:

• vms (list) – List of the IBMCLoud Bare metal machines objects to restart

• wait (bool) – If true, wait for VM to restart. False, otherwise.

• force (bool) – Force stop a VM. This option might cause data inconsistency or data loss.

restart_kubevirt_vms_by_stop_and_start_teardown()

Start the vms in paused and stopped state.

run_kubevirt_vm_cmd(**kwargs)

start_kubevirt_vms(vms, wait=True, start_vm_in_pause_state=False)

Start the kubevirt VMs

Parameters:

• vms (list) – List of the kubevirt VM objects to start

• wait (bool) – If true, wait for VM to start. False, otherwise.

• start_vm_in_pause_state (bool) – If True, it will start the VMs in a pause state.

stop_kubevirt_vms(vms, wait=True, force=False)

Stop the kubevirt VMs

Parameters:

• vms (list) – List of the kubevirt VM objects to stop

• wait (bool) – If true, wait for VM to stop. False, otherwise.

• force (bool) – Force stop a VM. This option might cause data inconsistency or data loss.

unpause_kubevirt_vms(vms, wait=True)

Unpause the kubevirt VMs

Parameters:

• vms (list) – List of the kubevirt VM objects to unpause

• wait (bool) – If true, wait for VM to start. False, otherwise.

wait_for_vms_status(vms, expected_status, timeout=300)

Wait for the VMs to be in the expected status

Parameters:

• vms (list) – The list of the VM objects

• expected_status (str) – The expected status

• timeout (int) – Time to wait for the VMs to reach the expected status

Raises:

TimeoutExpiredError – If the VMs didn’t reach the expected status in the given timeout

ocs_ci.utility.kubevirt_vm.get_vm_name(vm)

Get the virtual machine name

Parameters:

vm (dict) – The dictionary that represents the virtual machine

Returns:

The virtual machine name

Return type:

str

ocs_ci.utility.load_balancer module

Module that contains all operations related to load balancer in a cluster

class ocs_ci.utility.load_balancer.LoadBalancer(host=None, user=None, private_key=None)

Bases: object

A class that handles all operations related to load balancer

compact_mode_route_ingress_trafic_to_control_plane_nodes()

This is applicable only for compact mode deployment. Configure haproxy to route ingress traffic to control plane nodes.

modify_haproxy_service()

Modify haproxy service

reload_daemon()

Reload daemon-reload

remove_boostrap_in_proxy()

Removes bootstrap IP from haproxy.conf

remove_compute_node_in_proxy()

Removes compute node IP’s from haproxy.conf

rename_haproxy()

Rename haproxy configuration file from haproxy.conf to haproxy.cfg

rename_haproxy_conf_and_reload()

Rename haproxy config and restart haproxy service

restart_haproxy()

Restarts haproxy service

Returns:

True if successful restarts of haproxy, False otherwise

Return type:

bool

restart_service(service_name)

Restarts the given service

update_haproxy_with_nodes(nodes)

Parameters:

nodes (list) – List of nodes to update in haproxy

ocs_ci.utility.localstorage module

This module contains local-storage related methods

ocs_ci.utility.localstorage.check_local_volume_local_volume_set()

Function to check if Local-volume and Local volume set is present or not

Returns:

dict for localvolume and localvolumeset

Return type:

dict

ocs_ci.utility.localstorage.check_pvs_created(num_pvs_required)

Verify that exact number of PVs were created and are in the Available state

Parameters:

num_pvs_required (int) – number of PVs required

Raises:

AssertionError – if the number of PVs are not in the Available state

ocs_ci.utility.localstorage.fetch_all_device_paths()

Return all device paths inside worker nodes

Returns:

List containing all device paths

Return type:

list

ocs_ci.utility.localstorage.get_local_volume_cr()

Get localVolumeCR object

Returns:

Local Volume object handler

Return type:

local volume (obj)

ocs_ci.utility.localstorage.get_lso_channel()

Get the channel to use for installing the local storage operator

Returns:

local storage operator channel

Return type:

str

ocs_ci.utility.localstorage.get_new_device_paths(device_sets_required, osd_size_capacity_requested)

Get new device paths to add capacity over Baremetal cluster

Parameters:

• device_sets_required (int) – Count of device sets to be added

• osd_size_capacity_requested (int) – Requested OSD size capacity

Returns:

List containing added device paths

Return type:

list

ocs_ci.utility.lvm_utils module

ocs_ci.utility.lvm_utils.lvcreate(osd, lv_name, vg_name, size)

ocs_ci.utility.lvm_utils.make_partition(osd, device, start=None, end=None, gpt=False)

ocs_ci.utility.lvm_utils.osd_scenario1(osd, devices_dict, dmcrypt=False)

OSD scenario type1 generator :param osd: osd node :param devices_dict: dict of devices of the osd node supplied :param dmcrypt: False by default

Returns:

generated scenario, dmcrypt

ocs_ci.utility.lvm_utils.osd_scenario1_dmcrypt(osd, devices_dict)

OSD scenario type2 generator :param osd: osd node :param devices_dict: dict of devices of the osd node supplied :param dmcrypt: False by default

Returns:

generated scenario, dmcrypt(overridden to True)

ocs_ci.utility.lvm_utils.osd_scenario2(osd, devices_dict, dmcrypt=False)

OSD scenario type3 generator :param osd: osd node :param devices_dict: dict of devices of the osd node supplied :param dmcrypt: False by default

Returns:

generated scenario, dmcrypt

ocs_ci.utility.lvm_utils.osd_scenario2_dmcrypt(osd, devices_dict)

OSD scenario type4 generator :param osd: osd node :param devices_dict: dict of devices of the osd node supplied :param dmcrypt: False by default

Returns:

generated scenario, dmcrypt(overridden to True)

ocs_ci.utility.lvm_utils.osd_scenario3(osd, devices_dict, dmcrypt=False)

OSD scenario type5 generator :param osd: osd node :param devices_dict: dict of devices of the osd node supplied :param dmcrypt: False by default

Returns:

generated scenario, dmcrypt

ocs_ci.utility.lvm_utils.osd_scenario3_dmcrypt(osd, devices_dict)

OSD scenario type6 generator :param osd: osd node :param devices_dict: dict of devices of the osd node supplied :param dmcrypt: False by default

Returns:

generated scenario, dmcrypt(overridden to True)

ocs_ci.utility.lvm_utils.pvcreate(osd, devices)

ocs_ci.utility.lvm_utils.vgcreate(osd, vg_name, devices)

ocs_ci.utility.lvmo_utils module

ocs_ci.utility.lvmo_utils.delete_lvm_cluster()

Delete lvm cluster if exists

Raises:

execption if lvmcluster cant be deleted –

ocs_ci.utility.lvmo_utils.get_lvm_cluster_name()

Get LVM clustername Dynamically

Returns:

(str) lvm cluster name.

ocs_ci.utility.lvmo_utils.get_sno_blockdevices(node='sno-edge-0')

Gets list of storage devices by it’s names

Parameters:

node (str) – node name

Returns:

list of storage devices full-names (str)

Return type:

list

ocs_ci.utility.lvmo_utils.get_sno_disks_by_path(node='sno-edge-0')

Get list of storage devices by it’s path as listed on node

Parameters:

node (str) – node name

Returns:

list of storage devices full-path (str)

Return type:

list

ocs_ci.utility.lvmo_utils.lvmo_health_check(tries=20, delay=30)

Priodic check LVMO cluster health

Parameters:

• tries (int) – number of retries.

• delay (int) – delay between retries.

Returns:

True if cluster is healthy, raise exception otherwise.

Return type:

bool

ocs_ci.utility.lvmo_utils.lvmo_health_check_base()

Check health of lvm cluster by checking the following: lvm cluster cr status, lvm pods status

Returns:

True if all checks passed, raise exception otherwise

Return type:

bool

ocs_ci.utility.managedservice module

ocs_ci.utility.managedservice.add_header_footer_to_key(key)

This function will add header and footer to key (like: —–BEGIN RSA PRIVATE KEY—– —–END RSA PRIVATE KEY—– ) and return the key with header and footer.

Returns:

key string with header and footer

Return type:

string

ocs_ci.utility.managedservice.generate_onboarding_token(private_key: str = None, use_ticketgen_with_quota: bool = False, subject_role: str = None, storage_quota: int = None)

Generate Onboarding token for consumer cluster via following steps:

1. Download ticketgen.sh script from:

   https://raw.githubusercontent.com/red-hat-storage/ocs-operator/main/hack/ticketgen/ticketgen.sh

2. Save private key from AUTH[“managed_service”][“private_key”] to

   temporary file.

3. Run ticketgen.sh script to generate Onboarding token.

Important! The header and footer of the private key are rewritten to make the ticketgen.sh script work on older

openssl version OpenSSL 1.1.1k, which is the last version that was supported by CentOS 8 in the day of this comment

Parameters:

• private_key (str) – private key for Managed Service

• use_ticketgen_with_quota (bool) – If True, the ticketgen.sh script will be run with -q flag

• subject_role (str) – Role of the subject, this role has a default in ticketgen-with-quota.sh - ocs-client

• storage_quota (int) – Quota for the storage cluster

Raises:

• CommandFailed – In case the script ticketgen.sh fails.

• ConfigurationError – when AUTH[“managed_service”][“private_key”] not is not defined

Returns:

Onboarding token

Return type:

string

ocs_ci.utility.managedservice.get_storage_provider_endpoint(cluster)

Get get_storage_provider_endpoint

Parameters:

cluster (str) – cluster name

Returns:

value of storage provider endpoint

Return type:

str

ocs_ci.utility.managedservice.remove_header_footer_from_key(key)

This function will remove header and footer from key (like: —–BEGIN RSA PRIVATE KEY—– —–END RSA PRIVATE KEY—– ) and return the key on one line.

Returns:

one line key string without header and footer

Return type:

string

ocs_ci.utility.memory module

Module for memory related util functions.

class ocs_ci.utility.memory.MemoryMonitor(interval, function, args=None, kwargs=None)

Bases: Timer

class to implement threading.Timer running func in a background

run()

Method representing the thread’s activity.

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

ocs_ci.utility.memory.catch_empty_mem_df(df: DataFrame)

routine function to catch psutil failures and fill memory dataframe with failure markers, therefore we may see number of failures and ignore them on examination stage

ocs_ci.utility.memory.get_consumed_ram(proc: Process = psutil.Process(pid=1371, name='python', status='running', started='15:29:52'))

Get consumed RAM(rss) for the process rss is the Resident Set Size, which is the actual physical memory the process is using

ocs_ci.utility.memory.get_consumed_virt_mem(proc: Process = psutil.Process(pid=1371, name='python', status='running', started='15:29:52'))

Get consumed Virtual mem for the process vms is the Virtual Memory Size which is the virtual memory that process is using

! Important notice ! ‘vms’ will be larger than physical memory capacity: Shared libraries and frameworks are counted as part of the virtual memory for every application that uses them, e.g. if you have 100 processes running on a computer, and a 5 MB library used by all those processes, then that library is counted as 500 MB of virtual memory.

ocs_ci.utility.memory.get_peak_sum_mem() → tuple

get peak summarized memory stats for the test. Each test df file created anew. spikes defined per measurment (once in three seconds by default -> start_monitor_memory())

ocs_ci.utility.memory.mon: MemoryMonitor

ocs_ci.utility.memory.peak_mem_stats_human_readable(stat: <module 'ocs_ci.ocs.constants' from '/home/docs/checkouts/readthedocs.org/user_builds/ocs-ci/checkouts/latest/ocs_ci/ocs/constants.py'>, csv_path: str = None) → DataFrame

make peak mem stats dataframe human-readable dataframe columns = [name, proc_start, proc_end, rss_peak]

Parameters:

• stat (constants) – stat either ‘rss’ or ‘vms’ (constants.RAM | constants.VIRT)

• csv_path (str) – path to csv file with structure: index,pid,name,ts,rss,vms,status; will be ignored in case if df != None

Returns:

peak memory stats dataframe

Return type:

pd.DataFrame

ocs_ci.utility.memory.read_peak_mem_stats(stat: <module 'ocs_ci.ocs.constants' from '/home/docs/checkouts/readthedocs.org/user_builds/ocs-ci/checkouts/latest/ocs_ci/ocs/constants.py'>, df: ~pandas.core.frame.DataFrame = None, csv_path: str = None) → DataFrame

Read peak memory stats from Dataframe or csv file. Processes with stat above avg will be taken Table will be reduced to only processes with stat > avg(stat) if number of processes will be larger than 10

Parameters:

• stat (constants) – stat either ‘rss’ or ‘vms’ (constants.RAM | constants.VIRT)

• df (pd.DataFrame) – dataframe object with structure: index,pid,name,ts,rss,vms,status

• csv_path (str) – path to csv file with structure: index,pid,name,ts,rss,vms,status; will be ignored in case if df != None

Returns: pd.DataFrame similar to: name proc_start proc_end rss_peak 0 Google Chrome 2022-12-23 14:25:36 2022-12-23 14:27:32 156 MB 1 Google Chrome Helper (Renderer) 2022-12-23 14:25:39 2022-12-23 14:27:32 784 MB 2 Python 2022-12-23 14:25:22 2022-12-23 14:27:32 228 MB

ocs_ci.utility.memory.start_monitor_memory(interval: int = 3, create_csv: bool = False) → MemoryMonitor

Start memory monitor Timer process

Parameters:

• interval (int) – interval in sec to read measurements. Min interval is 2 sec

• create_csv (bool) – create csv during test run. With this option it is possible to upload file as artifact (to be done) or preserve csv file in the system

Returns:

monitor object MemoryMonitor(Timer)

Return type:

MemoryMonitor

ocs_ci.utility.memory.stop_monitor_memory(save_csv: bool = False) → tuple

Stop MemoryMonitor(Timer) and read memory stats

Parameters:

save_csv – saves csv temporarily, until main process is dead if save_csv = True; require create_csv=True at start_monitor_memory(…)

ocs_ci.utility.metadata_utils module

Module that contains all operations related to add metadata feature in a cluster

ocs_ci.utility.metadata_utils.available_subvolumes(sc_name, toolbox_pod, fs)

To fetch available subvolumes for cephfs or rbd

Parameters:

• sc_name (str) – storage class

• toolbox_pod (str) – ceph tool box pod

• fs (str) – file system

Returns:

subvolumes available for rbd or cephfs

Return type:

list

ocs_ci.utility.metadata_utils.check_setmetadata_availability(pod_obj)

Check if –setmetadata=true is present in CSI plugin pods’ args.

Parameters:

pod_obj (obj) – pod object

Returns:

True if –setmetadata=true is set on all CSI plugin pods, else False.

Return type:

bool

ocs_ci.utility.metadata_utils.created_subvolume(available_subvolumes, updated_subvolumes, sc_name)

To fetch created subvolume for cephfs or rbd

Parameters:

• available_subvolumes (list) – List of available subvolumes

• updated_subvolumes (list) – Updated list of subvolumes

• sc_name (str) – storage class

Returns:

name of subvolume created

Return type:

str

ocs_ci.utility.metadata_utils.enable_metadata(config_map_obj, pod_obj)

Enable CSI_ENABLE_METADATA through configmap or patch depending on OCS version.

Returns:

Cluster name if found, else None.

Return type:

str

ocs_ci.utility.metadata_utils.fetch_metadata(sc_name, fs, toolbox_pod, created_subvol, snapshot=False, available_subvolumes=None, updated_subvolumes=None)

To fetch metadata details created for cephfs or rbd

Parameters:

• sc_name (str) – storage class

• toolbox_pod (str) – ceph tool box pod

• fs (str) – file system

• created_subvol (str) – Created sub volume

• snapshot (bool) – snapshot or not

• available_subvolumes (list) – List of available subvolumes

• updated_subvolumes (list) – Updated list of subvolumes

Returns:

metadata details

Return type:

json

ocs_ci.utility.metadata_utils.patch_metadata(enable=True)

Patch CSI drivers to enable or disable metadata collection.

Parameters:

enable (bool) – Whether to enable or disable metadata.

ocs_ci.utility.metadata_utils.update_testdata_for_external_modes(sc_name, fs, external_mode=False)

Update the file sytem and storage class names for external mode clusters

Parameters:

• sc_name (str) – storage class

• fs (str) – file system

• external_mode (bool) – External mode or not

Returns:

storage class fs (str): file system

Return type:

sc_name (str)

ocs_ci.utility.metadata_utils.validate_metadata(metadata, clustername, pv_name=None, pvc_name=None, namespace=None, volumesnapshot_name=None, volumesnapshot_content=None)

To validate the metadata details

Parameters:

• metadata (json) – metadata details

• clustername (str) – cluster name

• pv_name (str) – name of the pv

• pvc_name (str) – name of the pvc

• namespace (str) – namespace

• volumesnapshot_name (str) – name of the volumesnapshot

• volumesnapshot_content (str) – volumesnapshot content

ocs_ci.utility.mirror_openshift module

ocs_ci.utility.mirror_openshift.prepare_mirror_openshift_credential_files()

This function is a context manager.

Prepare local files containing username and password for mirror.openshift.com:

data/mirror_openshift_user, data/mirror_openshift_password.

Those files are used as variable files for repo files (stored in /etc/yum/vars/ on the target RHEL nodes or pods).

Returns:

tuple containing two strings - file names for username file and password file

Return type:

(tuple)

Raises:

ConfigurationError – if mirror_openshift credentials are not provided in auth.yaml file

ocs_ci.utility.monitoring_tool module

ocs_ci.utility.monitoring_tool.check_go_version()

Check installed ‘go’ version

ocs_ci.utility.monitoring_tool.comparetool_deviation_check(first_file, second_file, deviation_list)

Function to run ‘comparetool’ and verify deviation_list accordingly to comparetool output

Parameters:

• first_file (Path) – path to the file, standing first in comparetool args to compare

• second_file (Path) – path to the file, standing second in comparetool args to compare

• deviation_list (list) – list of deviation strings expected to be in comparetool output

ocs_ci.utility.monitoring_tool.convert_yaml_file_to_json_file(file_path)

Util to convert yaml file to json file and replace an ext of the files

Parameters:

file_path (Path) – path to the file to convert

Returns:

path to the new file, yaml converted to json

Return type:

Path

ocs_ci.utility.multicluster module

All multicluster specific utility functions and classes can be here

class ocs_ci.utility.multicluster.MDRClusterUpgradeParametrize

Bases: MultiClusterUpgradeParametrize

This child class handles MDR upgrade scenario specific pytest parametrization

config_init()

class ocs_ci.utility.multicluster.MultiClusterUpgradeParametrize

Bases: object

This base class abstracts upgrade parametrization for multicluster scenarios: MDR, RDR and Managed service

MULTICLUSTER_UPGRADE_MARKERS = ['pre_upgrade', 'pre_ocp_upgrade', 'ocp_upgrade', 'post_ocp_upgrade', 'mco_upgrade', 'mce_upgrade', 'kubevirt_cluster_upgrade', 'dr_hub_upgrade', 'dr_cluster_operator_upgrade', 'acm_upgrade', 'pre_ocs_upgrade', 'ocs_upgrade', 'post_ocs_upgrade', 'post_upgrade']

config_init()

generate_config_index_map()

Generate config indexes for all the MDRs/RDR cluster roles ex: {“ActiveACM”: 0, “PassiveACM”: 2, “PrimaryODF”: 1, “SecondaryODF”: 3}

generate_pytest_parameters(metafunc, roles)

We will have to parametrize the test based on the MDR roles to which the test is applicable to, Parameters will be a tuple of (zone_rank, role_rank, config_index)

generate_role_ranks()

Based on current roles for MDR/RDR : ActiveACM:1, PassiveACM:1, PrimaryODF:2, SecondaryODF: 2 In case of RDR few runs might consider only one ACM where as in MDR we might have 2 ACMs. We will be adjusting the roles for RDR dynamically in the child class based on what type of run ( 1 ACM OR 2 ACMs)the user triggers

generate_role_to_param_tuple_map()

For each of the MDRs applicable roles store a tuple (zone_rank, role_rank, config_index) ex: {“ActiveACM”: (1, 1, 0), “PassiveACM”: (2, 1, 2), “PrimaryODF”: (1, 2, 1), “SecondarODF”: (2, 2, 3)}

generate_zone_ranks()

For each zone we would be generating the ranks, we will add the zone’s respective indexes to the base rank values which keeps the zone ranks apart and create spaces (for ranks) in between to accomodate other tests

generate_zone_role_map()

Generate a map of Cluster’s role vs zone in which clusters are located ex: {“ActiveACM”: ‘a’, “PassiveACM”: ‘b’, “PrimaryODF”: ‘a’}

get_all_acm_roles_to_param_tuple()

get_all_odf_roles_to_param_tuple()

get_all_roles_to_param_tuples()

get_dr_all_param_tuples(role)

get_pytest_params_tuple(role)

Get a tuple of parameters applicable to the given role For ex: if role is ‘ActiveACM’, then get a tuple which is applicable to that role. If the role is ‘all’ then we will get tuples of parameter for all the roles applicable Parmeter tuples looks like (zone_rank, role_rank, config_index) for a given role

get_roles(metafunc)

get_zone_info()

Get the list of participating zones

class ocs_ci.utility.multicluster.RDRClusterUpgradeParametrize

Bases: MultiClusterUpgradeParametrize

This child class handles RDR upgrade scenario specific pytest parametrization

UPGRADE_TEST_ORDER = {30: 1, 42: 3, 44: 4, 46: 5, 60: 2}

config_init()

reeval_upgrade_order(phase_order, zrank, role_rank)

Parameters:

• phase_order – The component order which is under upgrade for ex: ORDER_OCP_UPGRADE etc

• zrank – zone in which the cluster is present

• role_rank – Rank of the cluster based on role

Returns:

New value with which the test need to be order marked so that RDR specific

upgrade test order is preserved

Return type:

neworder (int)

ocs_ci.utility.multicluster.create_mce_catsrc()

Create MCE CatalogSource in the marketplace namespace 1. Create ImageDigestMirrorSet for ACM Deployment if not present 2. Create Konflux Catalogsource for MCE 3. Wait for Catalogsource to be in READY state

ocs_ci.utility.multicluster.get_multicluster_upgrade_parametrizer()

ocs_ci.utility.networking module

Module that contains network related functions

ocs_ci.utility.networking.add_data_replication_separation_to_cluster_data(cluster_data)

Update storage cluster YAML data with data replication separation if required.

This function retrieves the private IP and network information from a worker node using the ‘ip addr’ command and configures the storage cluster’s public network address range accordingly.

Parameters:

cluster_data (dict) – storage cluster YAML data

Returns:

updated storage storage cluster yaml

Return type:

dict

ocs_ci.utility.networking.annotate_worker_nodes_with_mon_ip()

Annotate worker nodes with annotation ‘network.rook.io/mon-ip: <IPAddress>’

This function retrieves the private IP address from each worker node using the ‘ip addr’ command and annotates the node with that IP address. Master nodes are also included to ensure all nodes hosting OCS/ODF components receive the annotation.

ocs_ci.utility.networking.create_drs_machine_config()

Create Machine Config that moves the second physical network to a bridge. This is done for HCP configuraion of data replication separation.

If the MachineConfig ‘99-br-storage-nmstate-worker’ already exists, this function will skip creation and return early.

ocs_ci.utility.networking.create_drs_nad(cluster_name)

Create NetworkAttachmentDefinition in namespace where the virt-launcher pods exist. This is done for HCP configuraion of data replication separation.

The namespace is constructed as f”clusters-{cluster_name}”. If the namespace doesn’t exist, it will be created automatically.

Parameters:

cluster_name (str) – cluster name used to construct the namespace on provider cluster

ocs_ci.utility.networking.get_excluded_network_cidrs()

Get network CIDR ranges that should be excluded when selecting node private IPs.

This includes: - clusterNetwork CIDRs from the OCP Network resource (pod network) - serviceNetwork CIDRs from the OCP Network resource (service network)

Returns:

List of ipaddress.IPv4Network objects representing excluded network CIDRs

Return type:

list

Raises:

UnavailableResourceException – If unable to retrieve Network resource

ocs_ci.utility.networking.get_node_private_ip(node_name)

Get the private IP address of a node using ip addr command.

This function executes ‘ip addr’ command on the node to retrieve the first globally routable IPv4 address that is not in the excluded networks (cluster network or service network). Link-local and other non-routable addresses are automatically filtered by selecting only addresses with “scope global”.

Parameters:

node_name (str) – Name of the node

Returns:

(interface_name (str), private_ip (str), prefix_length (str)) -

Interface name, IP address and network prefix length

Return type:

tuple

Raises:

• CommandFailed – If unable to retrieve IP address from the node

• UnavailableResourceException – If unable to retrieve Network resource

ocs_ci.utility.networking.get_pod_ips(pod_selector, namespace)

Get pod IPs for pods matching the selector in a given namespace. Only returns IPs for pods that are in Running state and not terminating.

Parameters:

• pod_selector (str) – Label selector to filter pods (e.g., “app=rook-ceph-osd”)

• namespace (str) – Namespace to search for pods

Returns:

Dictionary mapping pod names to their IP addresses

Return type:

dict

ocs_ci.utility.networking.is_ip_in_excluded_networks(ip_addr, excluded_cidrs)

Check if an IP address falls within any of the excluded network CIDR ranges.

Parameters:

• ip_addr (str) – IP address to check

• excluded_cidrs (list) – List of ipaddress.IPv4Network objects to check against

Returns:

True if IP is in any excluded network CIDR, False otherwise

Return type:

bool

ocs_ci.utility.nfs_utils module

Module that contains all operations related to nfs feature in a cluster

ocs_ci.utility.nfs_utils.check_cluster_resources_for_nfs(min_cpu=12, min_memory=32000000000)

Check if cluster has sufficient resources for NFS deployment.

For Provider-Client setups: - NFS runs on the Provider cluster, so only Provider needs to meet requirements - Client clusters just consume the distributed NFS StorageClass, so they skip this check

Parameters:

• min_cpu (int) – Minimum CPU cores per worker node (default: 12)

• min_memory (int) – Minimum memory in bytes per worker node (default: 32GB)

Returns:

True if cluster meets NFS resource requirements, False otherwise

Return type:

bool

ocs_ci.utility.nfs_utils.create_nfs_load_balancer_service(storage_cluster_obj)

Create the nfs loadbalancer service

Parameters:

storage_cluster_obj (obj) – storage cluster object

Returns:

host details

Return type:

str

ocs_ci.utility.nfs_utils.create_nfs_sc(sc_name_to_create, sc_name_to_copy='ocs-storagecluster-cephfs', retain_reclaim_policy=False, server=None)

This method is to create a new storageclass.

Parameters:

• sc_name_to_create (str) – name of the storageclass to create

• sc_name_to_copy (str) – name of the sc to copy

• retain_reclaim_policy (bool) – if true set retain reclaim policy

• server (str) – if server details passed set that value

Returns:

returns storageclass obj created

Return type:

new_sc(obj)

ocs_ci.utility.nfs_utils.delete_nfs_load_balancer_service(storage_cluster_obj)

Delete the nfs loadbalancer service and wait for it to be removed.

Parameters:

storage_cluster_obj (obj) – storage cluster object

ocs_ci.utility.nfs_utils.disable_nfs_service_from_provider(nfs_sc_obj, nfs_ganesha_pod_name)

This method is for disabling nfs feature from provider cluster

Parameters:

• nfs_ganesha_pod_name (str) – rook-ceph-nfs * pod name

• nfs_sc_obj (obj) – storage class object

ocs_ci.utility.nfs_utils.distribute_nfs_storage_class_to_all_consumers(nfs_sc)

This method is to distribute nfs storage class to Storage Consumers Function validates Storage Class is available on Client cluster and return combined result for all consumers.

Parameters:

nfs_sc (str) – nfs storage class name

Returns:

True if the nfs Storage Classes is distributed successfully to all consumers, False otherwise.

Return type:

bool

ocs_ci.utility.nfs_utils.fetch_nfs_server_details_on_client_cluster(default_server=False)

Fetch the NFS server endpoint configured on the client (consumer) cluster.

The NFS StorageClass is propagated asynchronously after the StorageConsumer spec is patched on the provider. This function retries until the StorageClass appears on the client cluster and its server parameter is populated.

Parameters:

default_server (bool) – If True, wait until the server parameter equals the default NFS service name (constants.NFS_DEFAULT_SERVICE_NAME). Use this when the external endpoint has been removed and the operator is expected to revert to the default. Default is False.

Returns:

NFS server endpoint (IP address or hostname) configured

in the NFS StorageClass.

Return type:

str

Raises:

TimeoutExpiredError – If the NFS StorageClass does not appear or the expected server value is not reached within the timeout.

ocs_ci.utility.nfs_utils.nfs_access_for_clients(nfs_sc)

This method is for client clusters to be able to access nfs

Parameters:

nfs_sc (str) – storage class name

Returns:

nfs-ganesha pod name str: host details

Return type:

str

ocs_ci.utility.nfs_utils.nfs_disable(storage_cluster_obj, config_map_obj, pod_obj, sc, nfs_ganesha_pod_name)

Disable nfs feature and ROOK_CSI_ENABLE_NFS Steps: 1:- Disable nfs feature for storage-cluster 2:- Disable ROOK_CSI_ENABLE_NFS via patch request 3:- Delete CephNFS, ocs nfs Service and the nfs StorageClass 4:- Wait untill nfs-ganesha pod deleted

Parameters:

• storage_cluster_obj (obj) – storage cluster object

• config_map_obj (obj) – config map object

• pod_obj (obj) – pod object

• sc (obj) – nfs storage class object

• nfs_ganesha_pod_name (str) – rook-ceph-nfs * pod name

ocs_ci.utility.nfs_utils.nfs_enable(storage_cluster_obj, config_map_obj, pod_obj, namespace)

Enable nfs feature and ROOK_CSI_ENABLE_NFS Steps: 1:- Enable nfs feature for storage-cluster 2:- Enable ROOK_CSI_ENABLE_NFS via patch request 3:- Check nfs-ganesha server is up and running 4:- Check csi-nfsplugin pods are up and running

Parameters:

• storage_cluster_obj (obj) – storage cluster object

• config_map_obj (obj) – config map object

• pod_obj (obj) – pod object

• namespace (str) – namespace name

Returns:

nfs-ganesha pod name

Return type:

str

ocs_ci.utility.nfs_utils.provisioner_selectors(nfs_plugins=False, cephfs_plugin=False)

This method returns the provisioner pod selectors

Parameters:

• nfs_plugins (bool) – if True returns nfs_plugin provisooner list

• cephfs_plugin (bool) – if True returns cephfs_plugin provisooner list

Returns:

list of nfs provisioner selectors cephfs_provisioner_list(list): list of cephfs provisioner selectors

Return type:

nfs_provisioner_list(list)

ocs_ci.utility.nfs_utils.remove_nfs_endpoint_details()

This method is to remove nfs external endpoint details if available

ocs_ci.utility.nfs_utils.remove_nfs_storage_class_from_all_consumers(nfs_sc)

Remove the NFS storage class from all ready StorageConsumer resources on the provider. This undoes the distribution done by distribute_nfs_storage_class_to_all_consumers().

Parameters:

nfs_sc (str) – NFS storage class name to remove

ocs_ci.utility.nfs_utils.skip_test_if_nfs_client_unavailable(nfs_client_ip)

Skip the tests if a valid nfs client ip is not available for outside mounts

Parameters:

nfs_client_ip (str) – nfs client ip address

ocs_ci.utility.nfs_utils.unmount(con, test_folder)

Unmount existing mount points

Parameters:

• con (obj) – connection object

• test_folder (str) – Mount path

ocs_ci.utility.nfs_utils.update_etc_hosts_on_nfs_client(con, hostname)

Resolve an NFS LB hostname from within the cluster and update /etc/hosts on the NFS client VM.

IBM Cloud VPC Load Balancer hostnames (*.lb.appdomain.cloud) are only resolvable from within the same VPC. When the NFS client VM is in a different VPC, DNS resolution fails and mounts hang. This function resolves the hostname by exec-ing on the node where the NFS pod runs, then writes the result into /etc/hosts on the client VM so mounts succeed.

This must be called after the LB service is created and after establishing the SSH connection to the NFS client VM. It is safe to call on every reconnect since it removes stale entries before writing new ones.

Parameters:

• con (Connection) – SSH connection to the NFS client VM

• hostname (str) – NFS LB hostname to resolve and add to /etc/hosts

ocs_ci.utility.nfs_utils.update_nfs_endpoint(host_details)

This method is to pass nfs external endpoint under storagecluster.spec.nfs

Parameters:

host_details (str) – host details

ocs_ci.utility.nfs_utils.wait_for_nfs_csi_config_on_client_cluster(timeout=300)

Wait until the NFS CSI configuration is populated in the ceph-csi-config ConfigMap on the client (consumer) cluster.

After distributing the NFS StorageClass via StorageConsumer.spec.storageClasses, the ocs-client-operator reconciles the ceph-csi-config ConfigMap asynchronously. Until the nfs section for a cluster entry is present, the NFS CSI provisioner cannot create PVCs.

Parameters:

timeout (int) – Maximum seconds to wait (default: 300).

Raises:

TimeoutExpiredError – If the NFS CSI config is not populated within timeout.

ocs_ci.utility.ocp_upgrade module

ocs_ci.utility.ocp_upgrade.pause_machinehealthcheck()

During the upgrade process, nodes in the cluster might become temporarily unavailable. In the case of worker nodes, the machine health check might identify such nodes as unhealthy and reboot them. To avoid rebooting such nodes, pause all the MachineHealthCheck resources before updating the cluster. This step is based on OCP documentation for OCP 4.9 and above.

ocs_ci.utility.ocp_upgrade.resume_machinehealthcheck(wait_for_mcp_complete=False, force_delete_pods=False)

Resume the machine health checks after updating the cluster. To resume the check, remove the pause annotation from the MachineHealthCheck resource.

Arg:

wait_for_mcp_complete(bool): If True run wait_for_machineconfigpool_status() force_delete_pods (bool): if True delete pods stuck at terminating forcefully

ocs_ci.utility.ocs_build module

This module is used to return latest OCS internal build for specified OCS version.

ocs_ci.utility.ocs_build.init_arg_parser()

Init argument parser.

Returns:

Parsed arguments

Return type:

object

ocs_ci.utility.ocs_build.main()

Main function

ocs_ci.utility.openshift_console module

class ocs_ci.utility.openshift_console.OpenshiftConsole(browser='chrome')

Bases: object

run_openshift_console(suite, env_vars=None, timeout=1500, log_suffix='')

Run openshift console suite

Parameters:

• suite (str) – openshift console suite to execute

• env_vars (dict) – env variables to expose for openshift console

• timeout (int) – timeout for test-gui.sh script

Raises:

OpenshiftConsoleSuiteNotDefined – if suite is not defined

setup_console_prereq()

Setup openshift console prerequisites

Raises:

UnsupportedBrowser – in case the browser is not supported

ocs_ci.utility.openshift_dedicated module

Module for interactions with Openshift Dedciated Cluster.

ocs_ci.utility.openshift_dedicated.create_cluster(cluster_name)

Create OCP cluster.

Parameters:

cluster_name (str) – Cluster name.

ocs_ci.utility.openshift_dedicated.destroy_cluster(cluster)

Destroy the cluster on Openshift Dedicated.

Parameters:

cluster (str) – Cluster name or ID.

ocs_ci.utility.openshift_dedicated.download_ocm_cli()

Method to download OCM cli

Returns:

path to the installer

Return type:

str

ocs_ci.utility.openshift_dedicated.get_admin_name(cluster)

Get admin name for the cluster. To get fresh credentials run first update_cached_credentials(cluster)

Parameters:

cluster (str) – Cluster name.

Returns:

Admin name.

Return type:

str

ocs_ci.utility.openshift_dedicated.get_cached_credentials(cluster)

Get cached credentials for the cluster or fetch if not cached.

Parameters:

cluster (str) – Cluster name.

Returns:

Cluster credentials.

Return type:

dict

ocs_ci.utility.openshift_dedicated.get_cluster_details(cluster)

Returns info about the cluster which is taken from the OCM command.

Parameters:

cluster (str) – Cluster name.

ocs_ci.utility.openshift_dedicated.get_credentials(cluster)

Get json with cluster credentials

Parameters:

cluster (str) – Cluster name.

Returns:

cluster credentials

Return type:

json

ocs_ci.utility.openshift_dedicated.get_kubeadmin_password(cluster, path)

Export password for kubeadmin to provided path. To get fresh credentials run first update_cached_credentials(cluster)

Parameters:

• cluster (str) – Cluster name.

• path (str) – Path where to create kubeadmin-password file.

ocs_ci.utility.openshift_dedicated.get_kubeconfig(cluster, path)

Export kubeconfig to provided path. To get fresh credentials run first update_cached_credentials(cluster)

Parameters:

• cluster (str) – Cluster name.

• path (str) – Path where to create kubeconfig file.

ocs_ci.utility.openshift_dedicated.list_cluster()

Returns info about the openshift dedciated clusters which is taken from the OCM command.

ocs_ci.utility.openshift_dedicated.login()

Login to OCM client

ocs_ci.utility.openshift_dedicated.update_cached_credentials(cluster)

Update cached credentials for the cluster.

Parameters:

cluster (str) – Cluster name.

ocs_ci.utility.operators module

Operators utility functions

class ocs_ci.utility.operators.IngressNodeFirewallOperator(create_catalog: bool = False)

Bases: Operator

verify_csv_status()

Verify the CSV status for the IngressNodeFirewall Operator deployment equals Succeeded

class ocs_ci.utility.operators.LocalStorageOperator(create_catalog: bool = False)

Bases: Operator

class ocs_ci.utility.operators.MetalLBOperator(create_catalog: bool = False)

Bases: Operator

class ocs_ci.utility.operators.NMStateOperator(create_catalog: bool = False)

Bases: Operator

count_nmstate_pods_running(count)

Count the pods for NMState Operator are running

Return type:

bool

create_nmstate_instance()

Create an instance of the nmstate Operator

verify_nmstate_csv_status()

Verify the CSV status for the nmstate Operator deployment equals Succeeded

verify_nmstate_pods_running()

Verify the pods for NMState Operator are running

class ocs_ci.utility.operators.OADPOperator(create_catalog: bool = False)

Bases: Operator

class ocs_ci.utility.operators.Operator(create_catalog: bool = False)

Bases: object

catalog_name: str = 'redhat-operators'

property catalog_selector

create_catalog()

Create a catalog for the operator

create_disconnected_catalog()

create_idms_for_unreleased_catalog()

create_namespace()

Create a namespace for the operator

create_operatorgroup()

Create an OperatorGroup for the operator.

If an OperatorGroup already exists in the namespace, skip creation to avoid OLM errors (TooManyOperatorGroups). This handles cases where platform operators (e.g., Fusion OdfManager) create OperatorGroups.

create_subscription()

Create a subscription for the operator

create_unreleased_catalog()

Create the unreleased catalog for the operator

deploy()

disconnected_required_packages: list[str] = []

Get the list of related images from the unreleased catalog image Example: ` oras discover --format         json quay.io/redhat-user-workloads/ocp-art-tenant/art-fbc:ocp__4.21__kubernetes-nmstate-rhel9-operator         | jq -r '.referrers[] | select(.artifactType == "application/vnd.konflux-ci.attached-artifact")         | .digest' | xargs -I {} oras pull         quay.io/redhat-user-workloads/ocp-art-tenant/art-fbc@{} && cat related-images.json         | jq -r '.[]' | sed 's/@sha256:.*//' `

get_channel()

Get the channel to use for installing the operator

Returns:

operator channel

Return type:

str

get_idms_data()

is_available()

Check if the operator is available

Returns:

True if the operator is available, False otherwise

Return type:

bool

name: str = None

namespace: str = None

operatorgroup_exists()

Check if an OperatorGroup already exists in the namespace.

This is important for platforms like Fusion Data Foundation where the platform controller (e.g., OdfManager) may create and manage the OperatorGroup automatically.

Returns:

True if at least one OperatorGroup exists, False otherwise

Return type:

bool

Raises:

CommandFailed – If checking for existing OperatorGroups fails

set_unreleased_catalog()

property unreleased_catalog_full_image

unreleased_catalog_image: str = 'quay.io/redhat-user-workloads/ocp-art-tenant/art-fbc'

unreleased_catalog_image_tag: str = None

property unreleased_catalog_name

property unreleased_idms_name

unreleased_images: list[str] = []

unreleased_mirror: str = 'quay.io/redhat-user-workloads/ocp-art-tenant/art-images-share'

ocs_ci.utility.pagerduty module

class ocs_ci.utility.pagerduty.PagerDutyAPI(token=None)

Bases: object

This is wrapper class for PagerDuty API: https://developer.pagerduty.com/api-reference/

In order to use the API, there must be set api_key. That is taken from AUTH/pagerduty/api_key in config.

check_incident_cleared(summary, measure_end_time, time_min=420, pagerduty_service_ids=None)

Check that all incidents with provided summary are cleared.

Parameters:

• summary (str) – Incident summary

• measure_end_time (int) – Timestamp of measurement end

• time_min (int) – Number of seconds to wait for incidents to be cleared since measurement end

• pagerduty_service_ids (list) – service ids used in incidents get query

create(resource, payload=None)

Post resources from PagerDuty API.

Parameters:

• resource (str) – Represents part of uri that specifies given resource

• payload (dict) – Provide parameters to POST API call.

Returns:

Response from PagerDuty api

Return type:

dict

delete(resource)

Delete resource from PagerDuty API.

Parameters:

resource (str) – Represents part of uri that specifies given resource

Returns:

Response from PagerDuty api

Return type:

dict

get(resource, payload=None)

Get resources from PagerDuty API.

Parameters:

• resource (str) – Represents part of uri that specifies given resource

• payload (dict) – Provide parameters to GET API call. e.g. for incidents resource this can be {“service_ids[]”: <id>, “since”: “2021-07-15T00:00:00Z”, “time_zone”: “UTC”}

Returns:

Response from PagerDuty api

Return type:

dict

get_escalation_policy_id(name)

Get account escalation policy from PagerDuty API by name.

Returns:

Escalation policy id

Return type:

str

get_integration_dict(vendor_name)

Get a structure prepared to be a payload for integration creation via API.

Parameters:

vendor_name (str) – Name of vendor that is used in integration

Returns:

Structure containing all data required to by PagerDuty API

in order to create an integration

Return type:

dict

get_service_dict()

Get a structure prepared to be a payload for service creation via API.

Returns:

Structure containing all data required to by PagerDuty API

in order to create a service

Return type:

dict

get_vendor_id(name)

Get id of vendor with provided name from PagerDuty API.

Parameters:

name (str) – Vendor name

Returns:

Vendor id

Return type:

str

wait_for_incident_cleared(summary, timeout=1200, sleep=5, pagerduty_service_ids=None)

Search for incident to be cleared.

Parameters:

• summary (str) – Incident summary

• sleep (int) – Number of seconds to sleep in between incidents search

• pagerduty_service_ids (list) – service ids used in incidents get query

Returns:

List of incident records

Return type:

list

ocs_ci.utility.pagerduty.check_incident_list(summary, urgency, incidents, status='triggered')

Check list of incidents that there are incidents with requested label in summary and specific urgency. If some incident is missing then this check fails.

Parameters:

• summary (str) – String that is part of incident summary

• urgency (str) – Incident urgency

• incidents (list) – List of incidents to check

• status (str) – Incident status

Returns:

List of incidents that match search requirements

Return type:

list

ocs_ci.utility.pagerduty.get_pagerduty_service_id()

Get the pagerduty service id in the current run

Returns:

The pagerduty service id in the current run

Return type:

str

ocs_ci.utility.pagerduty.set_pagerduty_faas_secret(integration_key)

Update managed-fusion-agent-config secret. This is valid only on Fusion aaS platform. managed-fusion-agent-config secret is expected to be present prior to the update.

Parameters:

integration_key (str) – Integration key taken from PagerDuty Prometheus integration

ocs_ci.utility.pagerduty.set_pagerduty_integration_secret(integration_key)

Update ocs-converged-pagerduty secret. This is valid only on ODF Managed Service. ocs-converged-pagerduty secret is expected to be present prior to the update.

Parameters:

integration_key (str) – Integration key taken from PagerDuty Prometheus integration

ocs_ci.utility.pgsql module

Module for interactions postgresql database.

class ocs_ci.utility.pgsql.PgsqlManager(username, password, keep_connection=False, database='postgres', host='localhost', port=5432)

Bases: object

class Cursor(conn)

Bases: object

Context manager which make sure the cursor is opened and closed.

close_conn()

Method to close the connection.

property conn

Connection property.

conn_closure()

Decorator which will make sure the connection is closed after call of method decorated with. This also depends on self.keep_connection which may not close the connection if set to True, but in case of some error or crash, it will make sure the connection is always properly closed.

create_database(db_name, extra_params)

Create database.

Parameters:

• db_name (string) – database name

• extra_params (string) – extra postgresql command to add after CREATE DATABASE db_name

delete_database(db_name)

Delete the database.

Parameters:

db_name (string) – database name

ocs_ci.utility.powernodes module

class ocs_ci.utility.powernodes.PowerNodes

Bases: object

Wrapper for PowerNodes

iskvm()

Verify if this environment is libvirt or kvm.

Returns:

True if this is kvm environment, False otherwise

Return type:

bool

restart_powernodes_machines(powernode_machines, timeout, wait, force=True)

Restart PowerNode Machines

Parameters:

• powernode_machines (list) – PowerNode objects

• timeout (int) – time in seconds to wait for node to reach ‘not ready’ state, and ‘ready’ state.

• wait (bool) – True if need to wait till the restarted node reaches timeout

• force (bool) – True for PowerNode ungraceful power off, False for graceful PowerNode shutdown - for future use

restart_powernodes_machines_powervs(powernode_machines, timeout, wait)

Restart PowerNode Machines

Parameters:

• powernode_machines (list) – PowerNode objects

• timeout (int) – time in seconds to wait for node to reach ‘not ready’ state, and ‘ready’ state.

• wait (bool) – True if need to wait till the restarted node reaches timeout

start_powernodes_machines(powernode_machines, timeout=900, wait=True, force=True)

Start PowerNode Machines

Parameters:

• powernode_machines (list) – List of PowerNode machines

• timeout (int) – time in seconds to wait for node to reach ‘not ready’ state, and ‘ready’ state.

• wait (bool) – Wait for PowerNodes to start - for future use

• force (bool) – True for PowerNode ungraceful power off, False for graceful PowerNode shutdown - for future use

start_powernodes_machines_powervs(powernode_machines, timeout=900, wait=True)

Start PowerNode Machines

Parameters:

• powernode_machines (list) – List of PowerNode machines

• timeout (int) – time in seconds to wait for node to reach ‘not ready’ state, and ‘ready’ state.

• wait (bool) – Wait for PowerNodes to start - for future use

stop_powernodes_machines(powernode_machines, timeout=900, wait=True, force=True)

Stop PowerNode Machines

Parameters:

• powernode_machines (list) – PowerNode objects

• timeout (int) – time in seconds to wait for node to reach ‘not ready’ state

• wait (bool) – True if need to wait till the restarted node reaches timeout - for future use

• force (bool) – True for PowerNode ungraceful power off, False for graceful PowerNode shutdown - for future use

Raises:

UnexpectedBehaviour – If PowerNode machine is still up

stop_powernodes_machines_powervs(powernode_machines, timeout=900, wait=True)

Stop PowerNode Machines - PowerVS

Parameters:

• powernode_machines (list) – PowerNodes objects

• timeout (int) – time in seconds to wait for node to reach ‘not ready’ state

• wait (bool) – True if need to wait till the restarted node reaches timeout - for future use

verify_machine_is_down(node)

Verify if PowerNode is completely powered off

Parameters:

node (object) – Node objects

Returns:

True if machine is down, False otherwise

Return type:

bool

ocs_ci.utility.prometheus module

class ocs_ci.utility.prometheus.PrometheusAPI(user=None, password=None, threading_lock=None, cluster_context=<class 'ocs_ci.framework.MultiClusterConfig.RunWithProviderConfigContextIfAvailable'>)

Bases: object

This is wrapper class for Prometheus API.

check_alert_cleared(label, measure_end_time, time_min=120)

Check that all alerts with provided label are cleared.

Parameters:

• label (str) – Alerts label

• measure_end_time (int) – Timestamp of measurement end

• time_min (int) – Number of seconds to wait for alert to be cleared since measurement end

generate_cert()

Generate CA certificate from kubeconfig for API.

TODO: find proper way how to generate/load cert files.

get(resource, payload=None, timeout=300)

Get alerts from Prometheus API.

Parameters:

• resource (str) – Represents part of uri that specifies given resource

• payload (dict) – Provide parameters to GET API call. e.g. for alerts resource this can be {‘silenced’: False, ‘inhibited’: False}

• timeout (int) – Number of seconds to wait for Prometheus endpoint to get available if it is not available

Returns:

Response from Prometheus alerts api

Return type:

requests.models.Response

prometheus_log(prometheus_alert_list)

Log all alerts from Prometheus API to list

Parameters:

prometheus_alert_list (list) – List to be populated with alerts

query(query, timestamp=None, timeout=None, validate=True, mute_logs=False, log_debug=False)

Perform Prometheus instant query. This is a simple wrapper over get() method with plumbing code for instant queries, additional validation and logging.

Parameters:

• query (str) – Prometheus expression query string.

• timestamp (str) – Evaluation timestamp (rfc3339 or unix timestamp). Optional.

• timeout (str) – Evaluation timeout in duration format. Optional.

• validate (bool) – Perform basic validation on the response. Optional, True is the default. Use False when you expect query to fail eg. during negative testing.

• mute_logs (bool) – True for muting the logs, False otherwise

• log_debug (bool) – True for logging in debug, False otherwise

Returns:

Result of the query (value(s) for a single timestamp)

Return type:

list

query_range(query, start, end, step, timeout=None, validate=True)

Perform Prometheus range query. This is a simple wrapper over get() method with plumbing code for range queries, additional validation and logging.

Parameters:

• query (str) – Prometheus expression query string.

• start (str) – start timestamp (rfc3339 or unix timestamp)

• end (str) – end timestamp (rfc3339 or unix timestamp)

• step (float) – Query resolution step width as float number of seconds.

• timeout (str) – Evaluation timeout in duration format. Optional.

• validate (bool) – Perform basic validation on the response. Optional, True is the default. Use False when you expect query to fail eg. during negative testing.

Returns:

result of the query

Return type:

list

refresh_connection()

Login into OCP, refresh endpoint and token.

verify_alerts_via_prometheus(expected_alerts, threading_lock)

Verify Alerts on prometheus

Parameters:

• expected_alerts (list) – list of alert names

• threading_lock (threading.Rlock) – Lock object to prevent simultaneous calls to ‘oc’

Returns:

True if expected_alerts exist, False otherwise

Return type:

bool

wait_for_alert(name, state=None, timeout=1200, sleep=5, min_count=1)

Search for alerts that have requested name and state.

Parameters:

• name (str) – Alert name

• state (str) – Alert state, if provided then there are searched alerts with provided state. If not provided then alerts are searched for absence of the alert. Loop that looks for alerts is broken when there are no alerts returned from API. This is done because API is not returning any alerts that are not in pending or firing state

• timeout (int) – Number of seconds for how long the alert should be searched

• sleep (int) – Number of seconds to sleep in between alert search

• min_count (int) – Minimum number of matching alerts required before breaking out of the polling loop (only applies when state is provided). Defaults to 1.

Returns:

List of alert records

Return type:

list

class ocs_ci.utility.prometheus.PrometheusAlertSubscriber(threading_lock, interval: float)

Bases: Timer

clear_alerts()

Clear alert list

get_alerts()

Get list of all alerts

prometheus_alert_list = []

run()

Run logging of all prometheus alerts.

! This method is called by Timer class, do not call it directly !

subscribe()

Start logging of all prometheus alerts

unsubscribe()

Stop logging of all prometheus alerts

ocs_ci.utility.prometheus.check_alert_list(label, msg, alerts, states, severity='warning', ignore_more_occurences=True, description=None, runbook=None)

Check list of alerts that there are alerts with requested label and message for each provided state. If some alert is missing then this check fails.

Parameters:

• label (str) – Alert label

• msg (str) – Alert message

• alerts (list) – List of alerts to check

• states (list) – List of states to check, order is important

• ignore_more_occurences (bool) – If true then there is checkced only occurence of alert with requested label, message and state but it is not checked if there is more of occurences than one.

• description (str) – Alert description

• runbook (str) – Alert’s runbook URL

ocs_ci.utility.prometheus.check_query_range_result_enum(result, good_values, bad_values=(), exp_metric_num=None, exp_delay=None, exp_good_time=None)

Check that result of range query matches given expectations. Useful for metrics which convey status (eg. ceph health, ceph_osd_up), so that you can assume that during a given period, it’s value should match given single (or tuple of) value(s).

Parameters:

• result (list) – Data from query_range() method.

• good_values (tuple) – Tuple of values considered good

• bad_values (tuple) – Tuple of values considered bad, indicating a problem (optional, use if you need to distinguish bad and invalid values)

• exp_metric_num (int) – expected number of data series in the result, optional (eg. for ceph_health_status this would be 1, but for something like ceph_osd_up this will be a number of OSDs in the cluster)

• exp_delay (int) – Number of seconds from the start of the query time range for which we should tolerate bad values. This is useful if you change cluster state and processing of this change is expected to take some time.

• exp_good_time (int) – Number of seconds during which we should see good values in the metrics data. When this time passess values can go bad (but can’t be invalid). If not specified, good values should be presend during the whole time.

Returns:

True if result matches given expectations, False otherwise

Return type:

bool

ocs_ci.utility.prometheus.check_query_range_result_limits(result, good_min, good_max, exp_metric_num=None, exp_delay=None, exp_good_time=None)

Check that result of range query matches given expectations. Useful for metrics which convey continuous value, eg. storage or cpu utilization.

Parameters:

• result (list) – Data from query_range() method.

• good_min (float) – Min. value which is considered good.

• good_max (float) – Max. value which is still considered as good.

• exp_metric_num (int) – expected number of data series in the result, optional (eg. for ceph_health_status this would be 1, but for something like ceph_osd_up this will be a number of OSDs in the cluster)

• exp_delay (int) – Number of seconds from the start of the query time range for which we should tolerate bad values. This is useful if you change cluster state and processing of this change is expected to take some time.

• exp_good_time (int) – Number of seconds during which we should see good values in the metrics data. When this time passess values can go bad (but can’t be invalid). If not specified, good values should be presend during the whole time.

Returns:

True if result matches given expectations, False otherwise

Return type:

bool

ocs_ci.utility.prometheus.check_query_range_result_viafunction(result, is_value_good, is_value_bad=<function <lambda>>, exp_metric_num=None, exp_delay=None, exp_good_time=None, is_float=False)

Check that result of range query matches expectations expressed via is_value_good (and optionally is_value_bad) functions, which takes a value and returns True if the value is good (or bad).

Parameters:

• result (list) – Data from query_range() method.

• is_value_good (function) – returns True for a good value

• is_value_bad (function) – returns True for a bad balue, indicating a problem (optional, use if you need to distinguish bad and invalid values)

• exp_metric_num (int) – expected number of data series in the result, optional (eg. for ceph_health_status this would be 1, but for something like ceph_osd_up this will be a number of OSDs in the cluster)

• exp_delay (int) – Number of seconds from the start of the query time range for which we should tolerate bad values. This is useful if you change cluster state and processing of this change is expected to take some time.

• exp_good_time (int) – Number of seconds during which we should see good values in the metrics data. When this time passess values can go bad (but can’t be invalid). If not specified, good values should be presend during the whole time.

• is_float (bool) – assume that the value is float, otherwise assume int

Returns:

True if result matches given expectations, False otherwise

Return type:

bool

ocs_ci.utility.prometheus.log_parsing_error(query, resp_content, ex)

Log an error raised during parsing of a prometheus query.

Parameters:

• query (dict) – Full specification of a prometheus query.

• resp_content (bytes) – Response from prometheus

• ex (Exception) – Exception raised during parsing of prometheus reply.

ocs_ci.utility.prometheus.validate_alert(threading_lock, alert_constant, message, description, runbook, severity='warning', state='pending', timeout=1200, sleep=None)

Wait for an alert and validate its properties.

Parameters:

• threading_lock – Threading lock object for thread-safe Prometheus API operations

• alert_constant (str) – Alert name constant

• message (str) – Expected alert message

• description (str) – Expected alert description

• runbook (str) – Expected runbook URL

• severity (str) – Expected severity

• state (str) – Alert state to wait for

• timeout (int) – Timeout in seconds to wait for the alert

• sleep (int) – Optional polling interval for alert wait

Returns:

True if alert is validated successfully, False otherwise

Return type:

bool

ocs_ci.utility.prometheus.validate_status(content)

Validate content data from Prometheus. If this fails, Prometheus instance or a query is so broken that test can’t be performed. We assume that Prometheus reports “success” even for queries which returns nothing.

Parameters:

content (dict) – data from Prometheus

Raises:

• TypeError – when content is not a dict

• ValueError – when status of the content is not success

ocs_ci.utility.prometheus.wait_for_alert_cleared(api, alert_name, timeout=600)

Wait for a Prometheus alert to disappear (no longer firing or pending).

Parameters:

• api (PrometheusAPI) – Prometheus API instance.

• alert_name (str) – Alert name to wait for clearing.

• timeout (int) – Seconds to wait for the alert to clear.

Raises:

AssertionError – If the alert is still present after timeout.

ocs_ci.utility.prometheus.wait_for_alert_firing(api, alert_name, timeout=600, expected_severity=None, expected_message_substr=None, min_count=1)

Wait for a Prometheus alert to reach the firing state and validate it.

Parameters:

• api (PrometheusAPI) – Prometheus API instance.

• alert_name (str) – Alert name to wait for.

• timeout (int) – Seconds to wait for the alert to fire.

• expected_severity (str | None) – If given, assert that alert["labels"]["severity"] equals this value.

• expected_message_substr (str | None) – If given, assert that this substring appears (case-insensitive) in alert["annotations"]["message"].

• min_count (int) – Minimum number of firing alert instances to wait for. Defaults to 1.

Returns:

All fired alert records (at least min_count).

Return type:

list[dict]

Raises:

AssertionError – If fewer than min_count alerts fire, or any validation fails.

ocs_ci.utility.proxy module

ocs_ci.utility.proxy.update_kubeconfig_with_proxy_url_for_client(kubeconfig)

If it is a proxy environment and client_http_proxy is defined, update each cluster configuration in kubeconfig with proxy-url parameter to redirect client access through proxy server

Parameters:

kubeconfig (str) – path to kubeconfig file

ocs_ci.utility.psiutils module

A module for all PSI-openstack related utilities

class ocs_ci.utility.psiutils.PSIUtils(psi_conf)

Bases: object

A class for handling PSI functionalities

attach_volume(vol, instance_id)

Attach the given volume to the specific PSI openstack instance

Parameters:

• vol (Volume) – cinder volume object

• instance_id (str) – uuid of the instance

check_expected_vol_status(vol, expected_state)

Check status of the volume and return true if it matches the expected state

Parameters:

• vol (cinderclient.volume) – Volume object for which state needs to be checked

• expected_state (str) – Expected state of the volume

Returns:

True if state is same as expected else False

Return type:

bool

check_vol_deleted(vol)

Check whether its delete

Parameters:

vol (cinderclient.Volume) – volume object

Returns:

True if deleted else False

Return type:

bool

create_volume(name, size=100, volume_type='tripleo', metadata=None)

A function to create openstack volumes

Parameters:

• name (str) – display name of the volume

• size (int) – of the volume in GBs

• volume_type (str) – type of volume to be created

• metadata (dict) – Any {k,v} to be associated with volume

Returns:

cinderclient.Client.Volumes

Return type:

Volume

detach_and_delete_vols(volumes)

Detach and delete volumes from the list

Parameters:

volumes (list) – of Volume objects

get_instances_with_pattern(pattern)

Get instances matching pattern

Parameters:

pattern (str) – Pattern for matching instance name note: we are looking for only ‘name’ param for pattern matching

Returns:

novaclient.base.ListWithMeta

get_volumes_with_tag(tag)

Get PSI volumes having this tag (k,v)

Parameters:

tag (dict) – of desired (k,v)

Returns:

of cinderclient.<vers>.volumes object

Return type:

list

ocs_ci.utility.reporting module

ocs_ci.utility.reporting.get_polarion_id(upgrade=False)

Determine the polarion_id of the deployment or upgrade

Parameters:

upgrade (bool) – get upgrade_id if true, else get deployment_id

Returns:

polarion_id of the deployment or upgrade

Return type:

str

ocs_ci.utility.reporting.get_rp_launch_attributes()

Retrieve information from the config to use as launch attributes in ReportPortal.

Returns:

ReportPortal launch attributes

Return type:

dict

ocs_ci.utility.reporting.get_rp_launch_description()

Construct and return the ReportPortal launch description.

Returns:

ReportPortal launch description

Return type:

(str)

ocs_ci.utility.reporting.get_rp_launch_name()

Construct and return the ReportPortal launch name :returns: ReportPortal launch name :rtype: str

ocs_ci.utility.reporting.update_live_must_gather_image()

Update live must gather image in the config.

ocs_ci.utility.resource_check module

Module for resource check which was created during test cases

ocs_ci.utility.resource_check.create_resource_dct()

Set the environment status and assign it into RESOURCE_DICT_TEST dictionary

ocs_ci.utility.resource_check.get_environment_status_after_execution(exclude_labels=None)

Set the environment status and assign it into ENV_STATUS_POST dictionary. In addition, check for any leftovers from test execution

Parameters:

exclude_labels (list) – App labels to ignore leftovers

Raises:

ResourceLeftoversException – In case there are leftovers in the environment after the test execution

ocs_ci.utility.retry module

ocs_ci.utility.retry.catch_exceptions(*exceptions)

Catch unhandled exception and log the exception. This wrapper is useful to catch the exception(s) and perform actions after the call to the function. This function is stored in retry.py module because it is related to retrying

Parameters:

*exceptions – One or more exception classes to catch.

ocs_ci.utility.retry.retry(exception_to_check, tries=4, delay=3, backoff=2, text_in_exception=None, func=None, max_timeout=14400, max_delay=600)

Retry calling the decorated function using exponential backoff, with a maximum timeout and maximum delay.

Parameters:

• exception_to_check – The exception to check. May be a tuple of exceptions to check.

• tries – Number of times to try (not retry) before giving up.

• delay – Initial delay between retries in seconds.

• backoff – Backoff multiplier e.g., value of 2 will double the delay each retry. If the delay is greater than max_delay after multiplier, than max_delay is used as delay

• text_in_exception – Retry only when text_in_exception is in the text of exception.

• func – Function for garbage collector.

• max_timeout – Maximum total time for retries in seconds (default: 4 hours).

• max_delay – Maximum delay between retries in seconds (default: 600 seconds or 10 minutes).

ocs_ci.utility.retry.retry_until_exception(exception_to_check, tries=4, delay=3, backoff=2, text_in_exception=None, func=None)

Retry calling the decorated function using exponential backoff until the exception occurs.

Parameters:

• exception_to_check – the exception to check. may be a tuple of exceptions to check

• tries – number of times to try (not retry) before giving up

• delay – initial delay between retries in seconds

• backoff – Backoff multiplier e.g., value of 2 will double the delay each retry. If the delay is greater than max_delay after multiplier, than max_delay is used as delay

• text_in_exception – Retry only when text_in_exception is in the text of exception

• func – function for garbage collector

ocs_ci.utility.retry.validate_retry_params(tries, delay, backoff, max_timeout, max_delay)

Validate retry parameters to ensure the total retry time does not exceed max_timeout.

Parameters:

• tries – Number of retry attempts.

• delay – Initial delay between retries in seconds.

• backoff – Multiplicative factor for backoff.

• max_timeout – Maximum total time for retries in seconds.

• max_delay – Maximum delay between retries in seconds.

Raises:

ValueError – If the retry configuration exceeds the maximum timeout.

ocs_ci.utility.rgwutils module

ocs_ci.utility.rgwutils.get_rgw_count(ocs_version, is_upgrade, version_before_upgrade)

Get RGW Count

RGW Count is 2 if:

OCS 4.5 unless upgraded from a prior version OCS 4.6 arbiter deployment

Otherwise, RGW Count is 1

Parameters:

• ocs_version (str, float) – OCS Version

• is_upgrade (bool) – If cluster was upgraded to current version

• version_before_upgrade (str, float) – OCS Version prior to upgrade

Returns:

RGW Count

Return type:

int

ocs_ci.utility.rhv module

Module for interactions with RHV Cluster

class ocs_ci.utility.rhv.RHV(url=None, username=None, password=None, insecure=True, ca_file=None)

Bases: object

Wrapper for RHV

add_disk(vm, size, disk_format='RAW', disk_interface='VIRTIO_SCSI', sparse=None, pass_discard=None, storage_domain_id=None, timeout=120)

Attaches disk to VM

Parameters:

• vm (types.Vm) – Vm instance

• size (int) – size of disk in GB

• disk_format (str) – underlying storage format of disks (default: “RAW”)

• disk_interface (str) – underlying storage interface of disks communication with controller (default: ‘VIRTIO_SCSI’)

• sparse (bool) – disk allocation policy. True for sparse, false for preallocated (default: None)

• pass_discard (bool) – True if the virtual machine passes discard commands to the storage, False otherwise (default: None)

• storage_domain_id (str) – A unique identifier for the storage domain

• timeout (int) – The timeout in seconds for disk status OK (default: 120)

add_disks(num_disks, vm, size, disk_format='RAW', disk_interface='VIRTIO_SCSI', sparse=None, pass_discard=None, storage_domain_id=None, timeout=120)

Adds multiple disks to the VM

Parameters:

• num_disks – number of disks to add

• vm (types.Vm) – Vm instance

• size (int) – size of disk in GB

• disk_format (str) – underlying storage format of disks (default: “RAW”)

• disk_interface (str) – underlying storage interface of disks communication with controller (default: ‘VIRTIO_SCSI’)

• sparse (bool) – disk allocation policy. True for sparse, false for preallocated (default: None)

• pass_discard (bool) – True if the virtual machine passes discard commands to the storage, False otherwise (default: None)

• storage_domain_id (str) – A unique identifier for the storage domain

• timeout (int) – The timeout in seconds for disk status OK (default: 120)

close_engine_connection(connection)

Releases the resources used by this connection.

create_engine_connection()

Create the connection to the engine server

Returns:

Connection to the engine server

Return type:

connection

property disks_service

Reference to the disks service

property engine_connection

Get the connection to the engine server

Returns:

Connection to the engine server

Return type:

connection

get_compute_vms()

Gets all compute VM’s

Returns:

list of compute Vm instances

Return type:

list

get_disk_attachment_service(vm_id, disk_id)

Get the service that manages a specific disk attachment

Parameters:

• vm_id (str) – unique identifier of the vm

• disk_id (str) – unique identifier of the disk

get_disk_attachments(vm_id)

Get the disks that are attached to the virtual machine

Parameters:

vm_id (str) – unique identifier of the vm

Returns:

list of disks attachments

Return type:

list

get_disk_attachments_service(vm_id)

Get the service that manages the disk attachments of the vm

Parameters:

vm_id (str) – unique identifier of the vm

get_disk_service(disk_id)

Get the disk service managing a specific disk

Parameters:

disk_id (str) – unique identifier of the disk

get_disks_by_pattern(pattern=None, case_sensitive=True, key='name', filter_by_cluster_name=True)

Get the list of disks by pattern. If not specified it will return all the disks, or will return the list filtered by the cluster name.

Parameters:

• pattern (str) – query string to restrict the returned disks

• case_sensitive (bool) – True to take case into account, False to ignore case (default: True)

• key (str) – Either ‘name’, or ‘id’ (default: name)

• filter_by_cluster_name (bool) – Will be used only if the ‘pattern’ param not specified. If True it filters the disks by the cluster name, else if False it returns all disks.

Returns:

list of disks

Return type:

list

get_rhv_vm_instance(vm_name)

Get RHV VM instance

Parameters:

vm_name – name of RHV VM

Returns:

Vm instance

Return type:

vm (types.Vm)

get_vm_names()

Returns:

list of VMs name

Return type:

list

get_vm_service(vm_id)

Get the service that manages a specific virtual machine

Parameters:

vm_id (str) – unique identifier of the vm

get_vm_status(vm)

Get the power status of RHV VM

Parameters:

vm (str) – RHV VM instance

Returns :

str: Power status of RHV VM

get_vms_by_pattern(pattern=None, case_sensitive=True, key='name', filter_by_cluster_name=True)

Get the list of virtual machines of the system by pattern. If not specified it will return all the vms, or will return the list filtered by the cluster name.

Parameters:

• pattern (str) – query string to restrict the returned vms

• case_sensitive (bool) – True to take case into account, False to ignore case (default: True)

• key (str) – Either ‘name’, or ‘id’ (default: name)

• filter_by_cluster_name (bool) – Will be used only if the ‘pattern’ param not specified. If True it filters the Vm by the cluster name, else if False it returns all Vm.

Returns:

list of Vm instances

Return type:

list

reboot_rhv_vms(vms, timeout=600, wait=True, force=True)

Reboot the RHV virtual machines

Parameters:

• vms (list) – Names of RHV vms

• timeout (int) – time in seconds to wait for VM to reboot

• wait (bool) – Wait for RHV VMs to reboot

• force (bool) – True to reboot vm forcibly False otherwise

remove_disk(vm, identifier, key='name', detach_only=True)

Removes the disk attachment. By default, only detach the disk from the virtual machine, but won’t remove it from the system, unless the detach_only parameter is False

Parameters:

• vm (types.Vm) – Vm instance

• identifier (str) – unique identifier of the disk Either disk name or disk id

• key (str) – Either ‘name’ or ‘id’ (default: name)

• detach_only (bool) – True to only detach the disk from the vm but not removed from the system, False otherwise (default: True)

remove_disks(vm, detach_only=True)

Removes all the extra disks for a VM

Parameters:

• vm (types.Vm) – Vm instance

• detach_only (bool) – True to only detach the disk from the vm but not removed from the system, False otherwise (default: True)

restart_rhv_vms_by_stop_and_start(vms, wait=True, force=False)

Stop and Start RHV virtual machines

Parameters:

• vms (list) – list of RHV vm instances

• force (bool) – True for non-graceful VM shutdown, False for graceful VM shutdown.

• wait (bool) – Wait for RHV VMs to start

start_rhv_vms(vms, wait=True, timeout=600)

Run the RHV virtual machines

Parameters:

• vms (list) – list of RHV vm instances

• wait (bool) – Wait for RHV VMs to start

• timeout (int) – time in seconds to wait for VM to reach ‘up’ status.

stop_rhv_vms(vms, timeout=600, force=False)

Shutdown the RHV virtual machines

Parameters:

• vms (list) – list of RHV vm instances

• force (bool) – True for non-graceful VM shutdown, False for graceful VM shutdown.

• timeout (int) – time in seconds to wait for VM to reach ‘down’ status.

property system_service

Reference to the root of the services tree

The returned value is an instance of the SystemService class

property vms_service

Reference to the virtual machines service

ocs_ci.utility.rosa module

Module for interactions with Openshift Dedciated Cluster.

ocs_ci.utility.rosa.appliance_mode_cluster(cluster_name)

Create appliance mode provider cluster

Parameters:

cluster_name (str) – Cluster name

ocs_ci.utility.rosa.create_account_roles(prefix='ManagedOpenShift')

Create the necessary account-wide roles and policies, including operator-specific policies.

Important: - Each cluster should have a unique prefix for its account roles, rather than using a common prefix across the entire account. - If multiple clusters are deployed with the same role prefix, deleting account roles during the cluster destruction stage could lead to: - Loss of Red Hat (RH) support. - Disruption of communication with worker nodes.

Ensure that role prefixes are uniquely assigned per cluster to maintain cluster integrity and supportability.

Parameters:

prefix (str) – role prefix

ocs_ci.utility.rosa.create_cluster(cluster_name, version_str, region)

Create OCP cluster.

Parameters:

• cluster_name (str) – Cluster name

• version_str (str) – cluster version

• region (str) – Cluster region

ocs_ci.utility.rosa.create_oidc_config()

Create OIDC config and wait for it to appear in the list ! In a very extreme case, other OIDC config can be created in the same time failing TimeoutSampler and raising TimeoutExpiredError exception

Returns:

OIDC config id

Return type:

str

Raises:

TimeoutExpiredError – If OIDC config is not created in time

ocs_ci.utility.rosa.create_oidc_provider(cluster)

Create the OpenID Connect (OIDC) provider that the Operators will use to authenticate

Parameters:

cluster (str) – cluster name or cluster id

ocs_ci.utility.rosa.create_operator_roles(cluster='', prefix='', oidc_config_id='', aws_account_id='', account_roles_prefix='')

Create the cluster-specific Operator IAM roles. The roles created include the relevant prefix for the cluster name

Parameters:

• cluster (str) – cluster name

• prefix (str) – role prefix

• oidc_config_id (str) – OIDC config id

• aws_account_id (str) – AWS account id

• account_roles_prefix (str) – account roles prefix

ocs_ci.utility.rosa.delete_account_roles(prefix)

Delete account roles ! Important to not delete account roles if there are any clusters in the account using this prefix

Parameters:

prefix (str) – role prefix

ocs_ci.utility.rosa.delete_odf_addon(cluster)

Delete ODF Managed Service addon from cluster.

Parameters:

cluster (str) – cluster name or cluster id

ocs_ci.utility.rosa.delete_oidc_config(oidc_config_id)

Delete OIDC config

Parameters:

oidc_config_id (str) – OIDC config id

ocs_ci.utility.rosa.delete_oidc_provider(cluster_name)

Delete oidc provider of the given cluster

Parameters:

cluster_name (str) – the cluster name

ocs_ci.utility.rosa.delete_operator_roles(prefix)

Delete operator roles with prefix

Parameters:

prefix (str) – prefix. Usually it is cluster name set during ‘rosa create operator-roles’ command

ocs_ci.utility.rosa.destroy_appliance_mode_cluster(cluster)

Delete rosa cluster if appliance mode

Parameters:

cluster – name of the cluster

Returns:

True if appliance mode and cluster delete initiated

else False

Return type:

bool

ocs_ci.utility.rosa.destroy_rosa_cluster(cluster, best_effort=True)

Delete rosa cluster

Parameters:

• cluster (str) – name of the cluster

• best_effort (bool) – If True (true), ignore errors and continue with the deletion of the cluster

ocs_ci.utility.rosa.download_rosa_cli()

Method to download OCM cli

Returns:

path to the installer

Return type:

str

ocs_ci.utility.rosa.edit_addon_installation(addon_param_key, addon_param_value, cluster_name=None, addon_name=None, wait=True)

Edit a specific parameter of the odf addon installation of a cluster

Parameters:

• addon_param_key (str) – The addon key param to modify

• addon_param_value (str) – The addon value param to modify

• cluster_name (str) – The cluster name. The default value is ‘config.ENV_DATA[“cluster_name”]’

• addon_name (str) – The addon name. The default value is ‘addon_name = config.ENV_DATA[“addon_name”]’

• wait (bool) – If true, wait for the addon to be ready. False, otherwise.

ocs_ci.utility.rosa.get_addon_info(cluster, addon_name)

Get line related to addon from rosa list addons command.

Parameters:

• cluster (str) – cluster name

• addon_name (str) – addon name

Returns:

line of the command for relevant addon. If not found, it returns None.

Return type:

str

ocs_ci.utility.rosa.get_associated_oidc_config_id(cluster_name)

Get the associated OIDC config id of the given cluster

Parameters:

cluster_name (str) – The cluster name

Returns:

The OIDC config id

Return type:

str

ocs_ci.utility.rosa.get_console_url(cluster_name)

Get the console URL of the given cluster

Parameters:

cluster_name (str) – The cluster name

Returns:

The console URL

Return type:

str

ocs_ci.utility.rosa.get_latest_oidc_config_id()

ocs_ci.utility.rosa.get_oidc_config_ids(latest=False)

Get OIDC config ids. If latest is True, return only the latest OIDC config id.

Parameters:

latest (bool) – If True, return only the latest OIDC config id

Returns:

List of OIDC config ids

Return type:

list

ocs_ci.utility.rosa.get_oidc_endpoint_url(oidc_config_id)

Get OIDC provider endpoint URL for the given OIDC config id

Parameters:

oidc_config_id (str) – OIDC config id

Returns:

OIDC provider id

Return type:

str

ocs_ci.utility.rosa.get_operator_roles_data(prefix)

Get the operator roles with the given prefix

Parameters:

prefix (str) – role prefix

Returns:

JSON data of operator roles

Return type:

dict

ocs_ci.utility.rosa.get_rosa_cluster_service_id(cluster)

Get service id of cluster

Parameters:

cluster (str) – cluster name

Returns:

service id of cluster. If not found, it returns None.

Return type:

str

ocs_ci.utility.rosa.get_rosa_service_details(cluster)

Returns info about the rosa service cluster.

Parameters:

cluster (str) – Cluster name.

ocs_ci.utility.rosa.get_rosa_version()

Entry point for getting ROSA versions with config initialization. Supports –ocsci-conf to load configuration files and –ocp-version to check specific version.

ocs_ci.utility.rosa.install_odf_addon(cluster)

Install ODF Managed Service addon to cluster.

Parameters:

cluster (str) – cluster name or cluster id

ocs_ci.utility.rosa.is_odf_addon_installed(cluster_name=None)

Check if the odf addon is installed

Parameters:

cluster_name (str) – The cluster name. The default value is ‘config.ENV_DATA[“cluster_name”]’

Returns:

True, if the odf addon is installed. False, otherwise

Return type:

bool

ocs_ci.utility.rosa.label_nodes(cluster_name, machinepool_id, labels, rewrite=False)

Label nodes of the given cluster. ! Important This method rewrites existing behavior of labeling nodes in the cluster, it appends the labels to the existing labels, but not rewrite them. This prevents the issue of accidental overwriting the existing labels.

Parameters:

• cluster_name (str) – The cluster name

• machinepool_id (str) – The machinepool id

• labels (str) – The labels to apply

• rewrite (bool) – If True, rewrite the labels. False, otherwise.

Returns:

The output of the command

Return type:

str

ocs_ci.utility.rosa.login()

Login to ROSA client

ocs_ci.utility.rosa.post_onboarding_verification()

Check that after onboarding consumer the relevant resources were created in the provider cluster and then switch back to the original cluster: 1. StorageConsumer with the correct id exists and all its CephResources are in Ready status 2. CephBlockPool and Subvolumegroup with the correct id are in Ready status

ocs_ci.utility.rosa.rosa_create_htpasswd_idp(htpasswd_path, cluster_name=None, idp_name='my_htpasswd')

Creates HTPasswd IDP from htpasswd file

Parameters:

• htpasswd_path (str) – Path to htpasswd file

• cluster_name (str) – Cluster name

• idp_name (str) – Name of the IDP

ocs_ci.utility.rosa.rosa_delete_htpasswd_idp(cluster_name=None, idp_name='my_htpasswd')

Deletes IDP

Parameters:

• cluster_name (str) – Cluster name

• idp_name (str) – Name of the IDP

ocs_ci.utility.rosa.rosa_list_idps(cluster_name=None)

List IDPs

Parameters:

cluster_name (str) – Cluster name

Returns:

Dictionary with IDP names as keys and IDP types as values

Return type:

dict

ocs_ci.utility.rosa.rosa_ocp_version_endpoint()

Endpoint for getting available OCP versions form ROSA.

ocs_ci.utility.rosa.upgrade_rosa_cluster(cluster_name, version)

Upgrade the ROSA cluster to the given version ! important ! rosa cli version drops error in case if –control-plane parameter is not used ! important ! upgrade is not performed automatically in case of ROSA clusters, especially in case of HCP; Upgrade is controlled by the Hive Operator; schedule depends on a Control Plane Queue

Parameters:

• cluster_name (str) – The cluster name

• version (str) – The version to upgrade the cluster

ocs_ci.utility.rosa.validate_ocp_version(version_str)

Validate the version whether given version is z-stream version available for ROSA.

Parameters:

version_str (str) – OCP version string

Returns:

True if given version is available in z-stream version for ROSA

else False

Return type:

bool

ocs_ci.utility.rosa.wait_console_url(cluster_name, timeout=600, sleep=10)

Wait for the console URL of the cluster to be ready

Parameters:

• cluster_name (str) – The cluster name

• timeout (int) – Timeout to wait for the console URL to be ready

• sleep (int) – Time in seconds to sleep between attempts

Returns:

The console URL

Return type:

str

Raises:

TimeoutExpiredError – In case the console URL is not ready in the given timeout

ocs_ci.utility.rosa.wait_for_addon_to_be_ready(cluster_name=None, addon_name=None, timeout=300, sleep=20)

Wait for the addon of a cluster to be ready

Parameters:

• cluster_name (str) – The cluster name. The default value is ‘config.ENV_DATA[“cluster_name”]’

• addon_name (str) – The addon name. The default value is ‘addon_name = config.ENV_DATA[“addon_name”]’

• timeout (int) – Timeout to wait for the addon to be ready

• sleep (int) – Time in seconds to sleep between attempts

Raises:

TimeoutExpiredError – In case the addon is not ready in the given timeout

ocs_ci.utility.rosa.wait_operator_roles(prefix, wait_minutes=10)

Wait for the operator roles to be created

Parameters:

• prefix (str) – role prefix

• wait_minutes (int) – Time in minutes to wait for operator roles to be created

Returns:

True if operator roles are created, False otherwise

Return type:

bool

ocs_ci.utility.secret module

ocs_ci.utility.secret.link_all_sa_and_secret(secret_name, namespace)

Link all service accounts in specified namespace with the secret for pulling of images.

Parameters:

• secret_name (str) – secret name

• namespace (str) – namespace name

ocs_ci.utility.secret.link_all_sa_and_secret_and_delete_pods(secret_name, namespace)

Link all service accounts in specified namespace with the secret for pulling of images.

Parameters:

• secret_name (str) – secret name

• namespace (str) – namespace name

ocs_ci.utility.secret.link_sa_and_secret(sa_name, secret_name, namespace)

Link service account and secret for pulling of images.

Parameters:

• sa_name (str) – service account name

• secret_name (str) – secret name

• namespace (str) – namespace name

ocs_ci.utility.service module

class ocs_ci.utility.service.KubeletService

Bases: Service

Kubelet Service class

The purpose of this class is to extend Service class to provide stop/start/restart etc operations on kubelet service. Since kubelet service stop and start operations involve changing status of OCP node objects, this class verifies the same.

restart(node, timeout)

Restart the kubelet service using parent service class. After that, ensures the corresponding OCP node is connectable and moves to Ready state.

Parameters:

• node (object) – Node objects

• timeout (int) – time in seconds to wait for service to stop.

start(node, timeout)

Start the kubelet service using parent service class. After that, ensures the corresponding OCP node is connectable and moves to Ready state.

Parameters:

• node (object) – Node objects

• timeout (int) – time in seconds to wait for service to stop.

stop(node, timeout)

Stop the kubelet service using parent service class. After that, ensures the corresponding OCP node moves to NotReady state.

Parameters:

• node (object) – Node objects

• timeout (int) – time in seconds to wait for service to stop.

class ocs_ci.utility.service.Service(service_name, force=True)

Bases: object

Generic Service class

The purpose of this class is to provide a support to perform start/stop/restart and status operations on a given service. The class is instantiated with required service name without ‘.service extension. Refer to KubeletService class in this module for an example on how to use these methods.

kill(node, timeout)

Kill the given service using systemctl.

Parameters:

• node (object) – Node objects

• timeout (int) – time in seconds to wait for service to be stopped.

restart(node, timeout)

Restart the given service using systemctl.

Parameters:

• node (object) – Node objects

• timeout (int) – time in seconds to wait for service to be started.

start(node, timeout)

Start the given service using systemctl.

Parameters:

• node (object) – Node objects

• timeout (int) – time in seconds to wait for service to start.

Raises:

UnexpectedBehaviour – If service on powerNode machine is still not up

status(node, timeout)

Get the status of the given service using systemctl.

Parameters:

• node (object) – Node objects

• timeout (int) – Future use.

Returns:

‘active’ or ‘inactive’ or ‘failed’, etc.

Return type:

(string)

stop(node, timeout)

Stop the given service using systemctl.

Parameters:

• node (object) – Node objects

• timeout (int) – time in seconds to wait for service to stop.

Raises:

UnexpectedBehaviour – If service on PowerNode machine is still up

verify_service(node, action)

Verify if PowerNode is completely powered off

Parameters:

• node (object) – Node objects

• action (string) – ACTIVE or INACTIVE or FAILED

Returns:

True if service state is reqested action, False otherwise

Return type:

bool

ocs_ci.utility.ssl_certs module

This module is used for generating custom SSL certificates.

class ocs_ci.utility.ssl_certs.CSR

Bases: X509Req

Wrapper class over crypto.X509Req used for customization.

class ocs_ci.utility.ssl_certs.Certificate(cn, sans=None)

Bases: object

Common certificate class.

Parameters:

• cn (str) – Certificate Common Name

• sans (list) – list of Subject Alternative Names (prefixed by the type like ‘DNS:’ or ‘IP:’)

property crt

property csr

property key

save_crt(path)

Save certificate to file

Parameters:

path (str) – path where to save certificate

save_csr(path)

Save certificate signing request to file

Parameters:

path (str) – path where to save certificate signing request

save_key(path)

Save certificate key to file

Parameters:

path (str) – path where to save certificate key

class ocs_ci.utility.ssl_certs.Key

Bases: PKey

Wrapper class over crypto.PKey used for customization.

class ocs_ci.utility.ssl_certs.LetsEncryptCertificate(dns_plugin, cn, sans=None)

Bases: Certificate

Generate custom certificate signed by Let’s Encrypt authority

Parameters:

• dns_plugin (str) – Certbot DNS Plugin name (default: ‘dns-route53’)

• cn (str) – Certificate Common Name

• sans (list) – list of Subject Alternative Names (prefixed by the type like ‘DNS:’ or ‘IP:’)

generate_csr()

Generate Certificate Signing Request for the certificate

generate_key()

Generate private key for the certificate

get_crt()

Get certificate from Let’s Encrypt

run_certbot()

class ocs_ci.utility.ssl_certs.OCSCertificate(signing_service, cn, sans=None)

Bases: Certificate

Generate custom certificate signed by the automatic signing certification authority

Parameters:

• signing_service (str) – URL of the automatic signing CA service

• cn (str) – Certificate Common Name

• sans (list) – list of Subject Alternative Names (prefixed by the type like ‘DNS:’ or ‘IP:’)

generate_csr()

Generate Certificate Signing Request for the certificate

generate_key()

Generate private key for the certificate

get_crt()

Use automatic signing CA service to sign the CSR

ocs_ci.utility.ssl_certs.configure_certs_arg_parser()

Init argument parser for configure-ssl-certs command.

Returns:

Parsed arguments

Return type:

object

ocs_ci.utility.ssl_certs.configure_certs_main()

Main function for configure-ssl-certs command. Configures custom SSL certificates for ingress and/or API.

ocs_ci.utility.ssl_certs.configure_custom_api_cert(skip_tls_verify=False, wait_for_machineconfigpool=True)

Configure custom SSL certificate for API. If the certificate doesn’t exists, generate new one signed by automatic certificate signing service.

Parameters:

• skip_tls_verify (bool) – True if allow skipping TLS verification

• wait_for_machineconfigpool (bool) – True if it should wait for machineConfigPool

Raises:

ConfigurationError – when some required parameter is not configured

ocs_ci.utility.ssl_certs.configure_custom_ingress_cert(skip_tls_verify=False, wait_for_machineconfigpool=True)

Configure custom SSL certificate for ingress. If the certificate doesn’t exists, generate new one signed by automatic certificate signing service.

Parameters:

• skip_tls_verify (bool) – True if allow skipping TLS verification

• wait_for_machineconfigpool (bool) – True if it should wait for machineConfigPool

Raises:

ConfigurationError – when some required parameter is not configured

ocs_ci.utility.ssl_certs.configure_ingress_and_api_certificates(skip_tls_verify=False)

Configure custom SSL certificate for ingress and API.

Parameters:

skip_tls_verify (bool) – True if allow skipping TLS verification

ocs_ci.utility.ssl_certs.configure_trusted_ca_bundle(ca_cert_path, skip_tls_verify=False)

Configure cluster-wide trusted CA bundle in Proxy object

Parameters:

• ca_cert_path (str) – path to CA Certificate(s) bundle file

• skip_tls_verify (bool) – True if allow skipping TLS verification

ocs_ci.utility.ssl_certs.create_ocs_ca_bundle(ca_cert_path, ca_bundle_name='ocs-ca-bundle', namespace='openshift-config', skip_tls_verify=False)

Create or update configmap object with ocs-ca-bundle

Parameters:

• ca_cert_path (str) – path to CA Certificate(s) bundle file

• ca_bundle_name (str) – name of the created or updated configmap object (default: ocs-ca-bundle)

• namespace (str) – namespace where to create the configmap (default: openshift-config)

• skip_tls_verify (bool) – True if allow skipping TLS verification

ocs_ci.utility.ssl_certs.get_cluster_info_from_cluster(skip_tls_verify=False)

Auto-detect cluster name and base domain from the running OpenShift cluster.

Parameters:

skip_tls_verify (bool) – Skip TLS verification

Returns:

(cluster_name, base_domain)

Return type:

tuple

Raises:

Exception – If unable to extract cluster information

ocs_ci.utility.ssl_certs.get_root_ca_cert()

If not available, download Root CA Certificate for custom Ingress certificate.

Returns

str: Path to Root CA Certificate

ocs_ci.utility.ssl_certs.init_arg_parser()

Init argument parser.

Returns:

Parsed arguments

Return type:

object

ocs_ci.utility.ssl_certs.main()

Main function for get-ssl-cert command

ocs_ci.utility.ssl_certs.update_kubeconfig_with_ca_cert(skip_tls_verify=False)

Update kubeconfig file with the kube-apiserver CA certificate authority data. This extracts the CA certificate from the cluster’s kube-apiserver-server-ca configmap and embeds it in the kubeconfig. This allows users who download the kubeconfig from a shared location to automatically trust the API server certificate without additional configuration.

Parameters:

skip_tls_verify (bool) – True if allow skipping TLS verification

ocs_ci.utility.storage_cluster_setup module

class ocs_ci.utility.storage_cluster_setup.StorageClusterSetup

Bases: object

Performs the setup of the StorageCluster for Data Foundation deployments

setup_storage_cluster()

ocs_ci.utility.storage_cluster_setup.create_external_pgsql_secret()

Creates secret for external PgSQL to be used by Noobaa

ocs_ci.utility.svt module

ocs_ci.utility.svt.svt_cleanup()

Removes clonned SVT project and virtual environemt and Projects Created while running SVT

Raises:

BaseException – In case any erros occured while removing project and ENV.

Returns:

True if No exceptions, False otherwise

Return type:

bool

ocs_ci.utility.svt.svt_cluster_loader(clusterload_file='/tmp/svt/openshift_scalability/config/master-vert.yaml')

ocs_ci.utility.svt.svt_create_venv_setup()

This function creates Virtual environemt for SVT project, and installs all the dependencies and activate the environment

ocs_ci.utility.svt.svt_project_clone()

This function clones the SVT project.

ocs_ci.utility.svt.svt_setup(iterations)

Creates and run svt workload

ocs_ci.utility.system module

ocs_ci.utility.system.is_path_empty(path)

Returns True if the given path does not contain any files, False otherwise.

Parameters:

path (str) – Path to be checked

ocs_ci.utility.templating module

class ocs_ci.utility.templating.Templating(base_path='/home/docs/checkouts/readthedocs.org/user_builds/ocs-ci/checkouts/latest/ocs_ci/templates')

Bases: object

Class which provides all functionality for templating

property base_path

Setter for self._base_path property

render_template(template_path, data)

Render a template with the given data.

Parameters:

• template_path (str) – location of the j2 template from the self._base_path

• data (dict) – the data to be formatted into the template

Returns: rendered template

ocs_ci.utility.templating.dump_data_to_json(data, json_file)

Dump data to json file

Parameters:

• data (dict) – dictionary with data to dump to the json file.

• json_file (str) – file path to json file

ocs_ci.utility.templating.dump_data_to_temp_yaml(data, temp_yaml)

Dump data to temporary yaml file

Parameters:

• data (dict or list) – dict or list (in case of multi_document) with data to dump to the yaml file.

• temp_yaml (str) – file path of yaml file

Returns:

dumped yaml data

Return type:

str

ocs_ci.utility.templating.dump_to_temp_yaml(src_file, dst_file, **kwargs)

Dump a jinja2 template file content into a yaml file

Args:

src_file (str): Template Yaml file path dst_file: the path to the destination Yaml file

ocs_ci.utility.templating.generate_yaml_from_jinja2_template_with_data(file_, **kwargs)

Generate yaml fron jinja2 yaml with processed data

Parameters:

file (str) – Template Yaml file path

Keyword Arguments:

attributes (All jinja2) –

Returns:

Generated from template file

Return type:

dict

Examples

generate_yaml_from_template(file_=’path/to/file/name’, pv_data_dict’)

ocs_ci.utility.templating.get_n_document_from_yaml(yaml_generator, index=0)

Returns n document from yaml generator loaded by load_yaml with multi_document = True.

Parameters:

• yaml_generator (generator) – Generator from yaml.safe_load_all

• index (int) – Index of document to return. (0 - 1st, 1 - 2nd document)

Returns:

Data from n document from yaml file.

Return type:

dict

Raises:

IndexError – In case that yaml generator doesn’t have such index.

ocs_ci.utility.templating.json_to_dict(json_file)

Converts JSON to dictionary format

Parameters:

json_file (str) – file path to json file

Returns:

JSON data in dictionary format

Return type:

dict

ocs_ci.utility.templating.load_config_data(data_path)

Loads YAML data from the specified path

Parameters:

data_path – location of the YAML data file

Returns: loaded YAML data

ocs_ci.utility.templating.load_yaml(file, multi_document=False)

Load yaml file (local or from URL) and convert it to dictionary

Parameters:

• file (str) – Path to the file or URL address

• multi_document (bool) – True if yaml contains more documents

Returns:

If multi_document == False, returns loaded data from yaml file

with one document.

generator: If multi_document == True, returns generator which each

iteration returns dict from one loaded document from a file.

Return type:

dict

ocs_ci.utility.templating.to_nice_yaml(a, indent=2, *args, **kw)

This is a j2 filter which allows you from dictionary to print nice human readable yaml.

Parameters:

• a (dict) – dictionary with data to print as yaml

• indent (int) – number of spaces for indent to be applied for whole dumped yaml. First line is not indented! (default: 2)

• *args – Other positional arguments which will be passed to yaml.dump

• *args – Other keywords arguments which will be passed to yaml.dump

Returns:

transformed yaml data in string format

Return type:

str

ocs_ci.utility.uninstall_openshift_logging module

Function to teardown the openshift-logging

ocs_ci.utility.uninstall_openshift_logging.check_pod_vanished(pod_names)

A function to check all the pods are vanished from the namespace

ocs_ci.utility.uninstall_openshift_logging.delete_logging_namespaces(force=False)

Deleting namespaces 1. UIPlugin 1. Openshift-cluster-observability-operator 2. Openshift-logging 3. Openshift-operators-redhat

ocs_ci.utility.uninstall_openshift_logging.uninstall_cluster_logging()

Function to uninstall cluster-logging from the cluster Deletes UI Plugin and projects “openshift-logging”, “openshift_cluster_observability_namespace” and “openshift-operators-redhat”

ocs_ci.utility.users module

class ocs_ci.utility.users.DevUser(username: str, password: str, secret_namespace: str, secret_name: str)

Bases: object

Data class representing a dev user with OpenShift and S3 credentials.

Fields:

username: OpenShift username for console login. password: OpenShift password for console login. secret_namespace: Namespace containing the S3 credentials secret. secret_name: Name of the secret containing S3 credentials.

password: str

secret_name: str

secret_namespace: str

username: str

ocs_ci.utility.users.add_htpasswd_user(username, password, htpasswd_path)

Create a new user credentials with provided username and password. These will be saved in file located on htpasswd_path. The file will be created if it doesn’t exist.

Parameters:

• username (str) – Name of a new user

• password (str) – Password for a new user

• htpasswd_path (str) – Path to httpasswd file

ocs_ci.utility.users.bind_user_to_noobaa_ui_role(username: str) → OCS

Bind a user to the noobaa-odf-ui ClusterRole.

Parameters:

username (str) – The OpenShift username to bind.

Returns:

The created ClusterRoleBinding resource.

Return type:

OCS

Raises:

CommandFailed – If resource creation fails.

ocs_ci.utility.users.create_htpasswd_idp()

Create OAuth identity provider of HTPasswd type. It uses htpass-secret secret as a source for list of users.

ocs_ci.utility.users.create_htpasswd_secret(htpasswd_path, replace=False)

Create or update htpass-secret secret from file located on htpasswd_path.

Parameters:

• htpasswd_path (str) – Path to httpasswd file

• replace (bool) – If secret already exists then this will replace it

ocs_ci.utility.users.create_noobaa_ui_clusterrole() → OCS

Create or update the noobaa-odf-ui ClusterRole for dev users.

This ClusterRole grants minimal read permissions to NooBaa resources, namespaces, and secrets, allowing users to access the Object Browser UI without full admin rights.

Uses ‘oc apply’ semantics to ensure the ClusterRole is always up-to-date.

Returns:

The ClusterRole resource.

Return type:

OCS

Raises:

CommandFailed – If resource apply fails.

ocs_ci.utility.users.delete_htpasswd_secret()

Delete HTPasswd secret.

ocs_ci.utility.users.delete_user_noobaa_ui_binding(username: str) → None

Delete the ClusterRoleBinding for a user.

Parameters:

username (str) – The OpenShift username.

ocs_ci.utility.users.get_server_url()

Get server URL.

Returns:

Server URL

Return type:

str

ocs_ci.utility.users.login(server_url, user, password)

Login to the cluster using provided username and password.

Parameters:

• server_url (str) – Server URL

• user (str) – Username

• password (str) – Password

ocs_ci.utility.users.logout()

Logout from the cluster.

ocs_ci.utility.users.user_factory(request, htpasswd_path)

Create a user factory.

Parameters:

• request (obj) – request fixture

• htpasswd_path (str) – Path to htpasswd file

Returns:

User factory function

Return type:

func

ocs_ci.utility.users.wait_for_rbac_propagation(username: str, resource: str, verb: str = 'list', timeout: int = 120, sleep: int = 10) → None

Wait for RBAC permissions to propagate for a user.

Uses ‘oc auth can-i –as=<username>’ to verify that the API server recognizes the user’s permissions before proceeding.

Parameters:

• username (str) – The OpenShift username to check.

• resource (str) – The resource to check (e.g., noobaas.noobaa.io).

• verb (str) – The verb to check.

• timeout (int) – Maximum wait time in seconds.

• sleep (int) – Interval between checks in seconds.

Raises:

TimeoutExpiredError – If permissions don’t propagate within timeout.

ocs_ci.utility.utils module

class ocs_ci.utility.utils.AZInfo

Bases: object

A class for getting different az numbers across calls

get_zone_number()

Increment current zone_number and perform modulus op to roll-on to next available number

Returns:

zone number index

Return type:

int

zone_number = 0

class ocs_ci.utility.utils.TimeoutIterator(timeout, sleep, func, func_args=None, func_kwargs=None)

Bases: TimeoutSampler

Wrapper of TimeoutSampler which separates parameters of the class itself and func arguments in __init__ method. Such way of passing function with parameters is used in python standard library.

This allows more explicit usage, which improves readability, eg.:

t1 = TimeoutIterator(timeout=60, sleep=5, func=foo, func_args=[bar])
t2 = TimeoutIterator(3600, sleep=10, func=foo, func_args=[bar])

class ocs_ci.utility.utils.TimeoutSampler(timeout, sleep, func, *func_args, **func_kwargs)

Bases: object

Samples the function output.

This is a generator object that at first yields the output of function func. After the yield, it either raises instance of timeout_exc_cls or sleeps sleep seconds.

Yielding the output allows you to handle every value as you wish.

Feel free to set the instance variables.

Parameters:

• timeout (int) – Timeout in seconds

• sleep (int) – Sleep interval in seconds

• func (function) – The function to sample

• func_args – Arguments for the function

• func_kwargs – Keyword arguments for the function

wait_for_func_status(result)

Get function and run it for given time until success or timeout. (using __iter__ function)

Parameters:

result (bool) – Expected result from func.

Examples:

sample = TimeoutSampler(
    timeout=60, sleep=1, func=some_func, func_arg1="1",
    func_arg2="2"
)
if not sample.wait_for_func_status(result=True):
    raise Exception

wait_for_func_value(value)

Implements common usecase of TimeoutSampler: waiting until func (given function) returns a given value.

Parameters:

value – Expected return value of func we are waiting for.

ocs_ci.utility.utils.activate_multiple_mdss(mds_nodes)

ocs_ci.utility.utils.add_chrony_to_ocp_deployment()

Create and Add necessary chrony resources

ocs_ci.utility.utils.add_info_about_mg_skips(soup)

ocs_ci.utility.utils.add_mem_stats(soup)

Add performance summary to the soup to print the table: columns = [‘TC name’, ‘Peak total RAM consumed’, ‘Peak total VMS consumed’, ‘RAM leak’]

ocs_ci.utility.utils.add_path_to_env_path(path)

Add path to the PATH environment variable (if not already there).

Parameters:

path (str) – Path which should be added to the PATH env. variable

ocs_ci.utility.utils.add_squad_analysis_to_email(session, soup)

Add squad analysis to the html test results used in email reporting

Parameters:

• session (obj) – Pytest session object

• soup (obj) – BeautifulSoup object of HTML Report data

ocs_ci.utility.utils.add_stage_cert()

Deploy stage certificate to the cluster.

ocs_ci.utility.utils.add_time_report_to_email(session, soup)

Takes the time report dictionary and converts it into HTML table

ocs_ci.utility.utils.allow_dir_fragmentation(mds_nodes)

ocs_ci.utility.utils.apply_oadp_workaround(namespace)

Apply the hostpath WA for IBMCloud’s https://access.redhat.com/articles/5456281#known-issues-with-cloud-providers-and-hyperscalers-18

Parameters:

namespace (str) – Namespace where oadp is running

ocs_ci.utility.utils.archive_ceph_crashes(toolbox_pod)

Archive all Ceph crashes

Parameters:

toolbox_pod (obj) – Ceph toolbox pod object

ocs_ci.utility.utils.auth_list(clients, mon_node)

ocs_ci.utility.utils.bin_xml_escape(arg)

Visually escape invalid XML characters.

For example, transforms ‘helloworld’ into ‘hello#x07world#x08’

Parameters:

arg (object) –

Returns:

string with escaped invalid characters

Return type:

str

ocs_ci.utility.utils.censor_values(data_to_censor)

This function censor string and numeric values in dictionary based on keys that match pattern defined in config_keys_patterns_to_censor in constants. It is performed recursively for nested dictionaries.

Parameters:

data_to_censor (dict) – Data to censor.

Returns:

filtered data

Return type:

dict

ocs_ci.utility.utils.ceph_crash_info_display(toolbox_pod)

Displays ceph crash information

Parameters:

toolbox_pod (obj) – Ceph toolbox pod object

ocs_ci.utility.utils.ceph_health_check(namespace=None, tries=20, delay=30, fix_ceph_health=False, update_jira=True, no_exception_if_jira_issue_updated=False)

Parameters:

• namespace (str) – Namespace of OCS (default: config.ENV_DATA[‘cluster_namespace’])

• tries (int) – Number of retries

• delay (int) – Delay in seconds between retries

• 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.

Returns:

ceph_health_check_base return value with default retries of 20,

delay of 30 seconds if default values are not changed via args.

Return type:

bool

ocs_ci.utility.utils.ceph_health_check_base(namespace=None, fix_ceph_health=False, update_jira=True, no_exception_if_jira_issue_updated=False)

Exec ceph health cmd on tools pod to determine health of cluster.

Parameters:

• namespace (str) – Namespace of OCS (default: config.ENV_DATA[‘cluster_namespace’])

• 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.

Raises:

• CephHealthException – If the ceph health returned is not HEALTH_OK

• CommandFailed – If the command to retrieve the tools pod name or the command to get ceph health returns a non-zero exit code

Returns:

True if HEALTH_OK

Return type:

boolean

ocs_ci.utility.utils.ceph_health_check_multi_storagecluster_external(tries=20, delay=30)

Check ceph health for multi-storagecluster external.

ocs_ci.utility.utils.ceph_health_multi_storagecluster_external_base()

Check ceph health for multi-storagecluster external implementation.

Returns:

True if cluster health is ok.

Return type:

bool

Raises:

CephHealthException – Incase ceph health is not ok.

ocs_ci.utility.utils.ceph_health_recover(health_status, namespace=None, update_jira=True, no_exception_if_jira_issue_updated=False)

Function which tries to recover ceph health to be HEALTH OK

Parameters:

• health_status (str) – Ceph health status

• namespace (str) – Namespace of OCS

• 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

Raises:

• CephHealthNotRecoveredException – When Ceph health was not recovered

• CephHealthRecoveredException – When Ceph health was recovered

ocs_ci.utility.utils.ceph_health_resolve_crash()

Fix ceph health issue with daemon crash

ocs_ci.utility.utils.ceph_health_resolve_mon_slow_ops(health_status)

Fix Ceph health issue with mon slow ops. Supports both ‘mon.e has slow ops’ and ‘daemons [mon.e,mon.f] have slow ops’

ocs_ci.utility.utils.ceph_health_resolve_network_partition(health_status)

Fix Ceph health issue with mon network partition.

ocs_ci.utility.utils.check_for_rhcos_images(url)

Check for rhcos images are present in given location

Parameters:

url (str) – rhcos_images url

Returns:

True if images present if not false

Return type:

(bool)

ocs_ci.utility.utils.check_if_executable_in_path(exec_name)

Checks whether an executable can be found in the $PATH

Parameters:

exec_name – Name of executable to look for

Returns:

Whether the executable was found

Return type:

Boolean

ocs_ci.utility.utils.check_timeout_reached(start_time, timeout, err_msg=None)

Check if timeout reached and if so raise the exception.

Parameters:

• start_time (time) – Star time of the operation.

• timeout (int) – Timeout in seconds.

• err_msg (str) – Error message for the exception.

Raises:

TimeoutException – In case the timeout reached.

ocs_ci.utility.utils.clean_up_pods_for_provider(node_type, max_retries=45, retry_delay_seconds=30)

Manually clean up pods if nodes get stuck in Ready,SchedulingDisabled during OCP upgrade. Checks machine-config-controller logs for eviction errors or completion messages.

The following pods may get stuck during cleanup in the hcpclusters namespace: - openshift-oauth-apiserver - oauth-openshift - openshift-apiserver - kube-apiserver - etcd-0 - virt-launcher pods

Parameters:

• node_type (str) – Type of nodes (e.g., “worker”, “master”) for which pods should be cleaned up.

• max_retries (int) – Maximum number of times to check nodes before giving up.

• retry_delay_seconds (int) – Seconds to wait between checks if no progress is made.

ocs_ci.utility.utils.clone_notify()

Repository contains the source code of notify tool, which is a python3 based tool wrapped by a container used to configure Ceph Bucket Notifications

Returns:

Path location of the notify code

Return type:

notify_path (str)

ocs_ci.utility.utils.clone_repo(url, location, tmp_repo=False, branch='master', to_checkout=None, clone_type='shallow', force_checkout=False)

Clone a repository or checkout latest changes if it already exists at

specified location.

Parameters:

• url (str) – location of the repository to clone

• location (str) – path where the repository will be cloned to

• tmp_repo (bool) – temporary repo, means it will be copied to temp path, to ‘location’

• branch (str) – branch name to checkout

• to_checkout (str) – commit id or tag to checkout

• clone_type (str) – type of clone (shallow, blobless, treeless and normal) By default, shallow clone will be used. For normal clone use clone_type as “normal”.

• force_checkout (bool) – True for force checkout to branch. force checkout will ignore the unmerged entries.

Raises:

UnknownCloneTypeException – In case of incorrect clone_type is used

ocs_ci.utility.utils.compare_dictionaries(dict1: dict, dict2: dict, known_different_keys: list, tolerance: int = 10) → dict

Compares two dictionaries and returns a dictionary with the keys that have different values, but allow a tolerance between this values.

Parameters:

• dict1 (dict) – dictionary to compare.

• dict2 (dict) – dictionary to compare.

• known_different_keys (list) – keys to ignore from the comparison.

• tolerance (int) – level of tolerance by precentage. Defaults to 10.

Returns:

difrerences between the two dictionaries.

Return type:

dict

ocs_ci.utility.utils.config_to_string(config)

Convert ConfigParser object to string in INI format.

Parameters:

config (obj) – ConfigParser object

Returns:

Config in one string

Return type:

str

ocs_ci.utility.utils.configure_chrony_and_wait_for_machineconfig_status(node_type='worker', timeout=900)

Configure chrony on the nodes

Parameters:

• node_type (str) – The node type to configure chrony e.g: worker, master and all if we want to configure on all nodes

• timeout (int) – Time in seconds to wait

ocs_ci.utility.utils.convert_bytes_to_unit(bytes_to_convert)

Convert bytes to bigger units like Kb, Mb, Gb or Tb.

Parameters:

bytes_to_convert (str) – The bytes to convert.

Returns:

The converted bytes as biggest unit possible

Return type:

str

ocs_ci.utility.utils.convert_device_size(unformatted_size, units_to_covert_to, convert_size=1000)

Convert a string representing a size to an int according to the given units to convert to

Parameters:

• unformatted_size (str) – The size to convert (i.e, ‘1Gi’/’100Mi’) acceptable units are: “Ti”, “Gi”, “Mi”, “Ki”, “Bi”

• units_to_covert_to (str) – The units to convert the size to (i.e, TB/GB/MB) acceptable units are: “TB”, “GB”, “MB”, “KB”, “BY”

• convert_size (int) – set convert by 1000 or 1024

Returns:

The converted size

Return type:

int

ocs_ci.utility.utils.convert_hostnames_to_ips(hostnames)

Gets the IP’s from hostname with FQDN

Parameters:

hostnames (list) – List of host names with FQDN

Returns:

Host IP’s

Return type:

list

ocs_ci.utility.utils.convert_yaml2tfvars(yaml)

Converts yaml file to tfvars. It creates the tfvars with the same filename in the required format which is used for deployment.

Parameters:

yaml (str) – File path to yaml

Returns:

File path to tfvars

Return type:

str

ocs_ci.utility.utils.create_ceph_health_cmd(namespace)

Forms the ceph health command

Parameters:

namespace (str) – Namespace of OCS

Returns:

ceph health command

Return type:

str

ocs_ci.utility.utils.create_config_ini_file(params)

This function will create a config ini file for the params given.

Use case: can be used in external-cluster-exporter script –config-file argument

Parameters:

params (str) – Parameter to pass to be converted into config.ini file.

Returns:

Path to the config.ini file.

Return type:

str

ocs_ci.utility.utils.create_directory_path(path)

Creates directory if path doesn’t exists

ocs_ci.utility.utils.create_kubeconfig(kubeconfig_path)

Create kubeconfig if doesn’t exist and OCP url and kubeadmin password is provided

Parameters:

kubeconfig_path (str) – kubeconfig file location

ocs_ci.utility.utils.create_rhelpod(namespace, pod_name, rhel_version=8, timeout=300)

Creates the RHEL pod

Parameters:

• namespace (str) – Namespace to create RHEL pod

• pod_name (str) – Pod name

• rhel_version (int) – RHEL version to be used for ansible

• timeout (int) – wait time for RHEL pod to be in Running state

Returns:

Pod instance for RHEL

Return type:

pod

ocs_ci.utility.utils.create_stats_dir()

create directory for Test run performance stats

ocs_ci.utility.utils.custom_ceph_config(suite_config, custom_config, custom_config_file)

Combines and returns custom configuration overrides for ceph. Hierarchy is as follows:

custom_config > custom_config_file > suite_config

Parameters:

• suite_config – ceph_conf_overrides that currently exist in the test suite

• custom_config – custom config args provided by the cli (these all go to the global scope)

• custom_config_file – path to custom config yaml file provided by the cli

Returns

New value to be used for ceph_conf_overrides in test config

ocs_ci.utility.utils.date_in_minimal_format()

Get the current date in a minimal format, such as 61024 for 6 of October 2024. Suitable to add to resource names.

Returns:

The current date in a minimal

Return type:

str

ocs_ci.utility.utils.decode(encoded_message)

Decodes the message in base64

Parameters:

encoded_message (str) – encoded message

Returns:

decoded message

Return type:

str

ocs_ci.utility.utils.decompose_html_attributes(soup, attributes)

Decomposes the given html attributes

Parameters:

• soup (obj) – BeautifulSoup object

• attributes (list) – attributes to decompose

Returns: None

ocs_ci.utility.utils.delete_dir(dir_name)

Deletes the directory

Parameters:

dir_name (str) – Directory path to delete

ocs_ci.utility.utils.delete_file(file_name)

Delete file_name

Parameters:

file_name (str) – Path to the file you want to delete

ocs_ci.utility.utils.destroy_cluster(installer, cluster_path, log_level='DEBUG')

Destroy OCP cluster specific

Parameters:

• installer (str) – The path to the installer binary

• cluster_path (str) – The path of the cluster

• log_level (str) – log level openshift-installer (default: DEBUG)

ocs_ci.utility.utils.disable_huge_pages()

Removes huge pages

ocs_ci.utility.utils.download_file(url, filename, **kwargs)

! Deprecated, use download_with_retries instead !

Download a file from a specified url

Parameters:

• url (str) – URL of the file to download

• filename (str) – Name of the file to write the download to

• kwargs (dict) – additional keyword arguments passed to requests.get(…)

ocs_ci.utility.utils.download_file_from_git_repo(git_repo_url, path_to_file_in_git, filename)

Download a file from a specified git repository

Parameters:

• git_repo_url (str) – The git repository url

• path_to_file_in_git (str) – Path to the file to download in git repository

• filename (str) – Name of the file to write the download to

ocs_ci.utility.utils.download_ga_oc_client_as_bootstrap(bin_dir, target_version)

Download a GA oc client as bootstrap to enable extraction from pullSpec.

This function downloads a GA version of the oc client (potentially N-1 or N-2) to use as a bootstrap client for extracting the correct nightly version from the release image pullSpec.

Parameters:

• bin_dir (str) – Directory to download the client to

• target_version (str) – Target version to determine which GA version to use

Returns:

True if successfully downloaded a bootstrap client, False otherwise

Return type:

bool

ocs_ci.utility.utils.download_with_retries(url, filename, max_retries=3)

Download file with retries and proper error handling

Parameters:

• url (str) – URL of the file to download

• filename (str) – Path where to save the downloaded file

• max_retries (int) – Maximum number of retries for downloading the file

Returns:

Path to the downloaded file if successful, None otherwise

Return type:

str

ocs_ci.utility.utils.dump_config_to_file(file_path)

Dump the config to the yaml file with censored secret values.

Parameters:

file_path (str) – Path to file where to write the configuration.

ocs_ci.utility.utils.email_reports(session)

Email results of test run

ocs_ci.utility.utils.enable_huge_pages()

Applies huge pages

ocs_ci.utility.utils.encode(message)

Encodes the message in base64

Parameters:

message (str/list) – message to encode

Returns:

encoded message in base64

Return type:

str

ocs_ci.utility.utils.ensure_nightly_build_availability(build_url)

ocs_ci.utility.utils.exec_cmd(cmd, secrets=None, timeout=600, ignore_error=False, threading_lock=None, silent=False, cluster_config=None, lock_timeout=7200, output_file=None, **kwargs)

Run an arbitrary command locally

If the command is grep and matching pattern is not found, then this function returns “command terminated with exit code 1” in stderr.

Parameters:

• cmd (str) – command to run

• secrets (list) – A list of secrets to be masked with asterisks This kwarg is popped in order to not interfere with subprocess.run(**kwargs)

• timeout (int) – Timeout for the command, defaults to 600 seconds.

• ignore_error (bool) – True if ignore non zero return code and do not raise the exception.

• threading_lock (threading.RLock) – threading.RLock object that is used for handling concurrent oc commands

• silent (bool) – If True will silent errors from the server, default false

• cluster_config (MultiClusterConfig) – In case of multicluster environment this object will be non-null

• lock_timeout (int) – maximum timeout to wait for lock to prevent deadlocks (default 2 hours)

• output_file (str) – path where to write output of stderr from command - apply only when silent mode is True

Raises:

CommandFailed – In case the command execution fails

Returns:

(CompletedProcess) A CompletedProcess object of the command that was executed CompletedProcess attributes: args: The list or str args passed to run(). returncode (str): The exit code of the process, negative for signals. stdout (str): The standard output (None if not captured). stderr (str): The standard error (None if not captured).

ocs_ci.utility.utils.exec_nb_db_query(query)

Send a psql query to the Noobaa DB

Example usage:

exec_nb_db_query(“SELECT data ->> ‘key’ FROM objectmds;”)

Parameters:

query (str) – The query to send

Returns:

The query result rows

Return type:

list of str

Raises:

ResourceNotFoundError – If no NooBaa DB pod is found

ocs_ci.utility.utils.expose_ocp_version(version)

This helper function exposes latest nightly version or GA version of OCP. When the version string ends with .nightly (e.g. 4.2.0-0.nightly) it will expose the version to latest accepted OCP build (e.g. 4.2.0-0.nightly-2019-08-08-103722) If the version ends with -ga than it will find the latest GA OCP version and will expose 4.2-ga to for example 4.2.22.

Parameters:

version (str) – Verison of OCP

Returns:

Version of OCP exposed to full version if latest nighly passed

Return type:

str

ocs_ci.utility.utils.expose_ocp_version_safe(version) → str

This helper function exposes latest nightly version or GA version of OCP.

ocs_ci.utility.utils.extract_image_urls(string_data)

Extract all image URLs from the string.

Parameters:

string_data (str) – The string data that contains the image URLs to extract.

Returns:

List of image URLs found in the string, or an empty list if none are found.

Return type:

list

ocs_ci.utility.utils.extract_ocp_binary_from_image(binary, image, bin_dir)

Extract binary (oc client or openshift installer) from custom OCP image

Parameters:

• binary (str) – type of binary (oc or openshift-install)

• image (str) – image URL

• bin_dir (str) – path to bin folder where to extract the binary

ocs_ci.utility.utils.file_locking(client)

ocs_ci.utility.utils.filter_out_emojis(plaintext)

Filter out emojis from a string

Parameters:

string (str) – String to filter out emojis from

Returns:

Filtered string

Return type:

str

ocs_ci.utility.utils.filter_unrepresentable_values(data_to_filter)

This function filter values in dictionary or list which are not possible convert to yaml (e.g. objects), to prevent following error raised from yaml.safe_dump yaml.representer.RepresenterError(“cannot represent an object”,…) It is performed recursively for nested dictionaries or lists.

Parameters:

data_to_filter (dict|list|tuple) – Data to censor.

Returns:

filtered data

Return type:

dict

ocs_ci.utility.utils.fuse_client_io(client, mounting_dir)

ocs_ci.utility.utils.fuse_client_md5(fuse_clients, md5sum_list1)

ocs_ci.utility.utils.fuse_mount(fuse_clients, mounting_dir)

ocs_ci.utility.utils.generate_folder_with_files(num_files: int = 300, max_workers: int = 4) → tuple[str, Iterator[str]]

Generates a random folder and returns a generator that creates random text files on demand. Uses ThreadPoolExecutor for improved performance.

Parameters:

• num_files (int) – Maximum number of files to generate. Defaults to 300.

• max_workers (int) – Maximum number of threads to use. Defaults to 4.

Returns:

Tuple containing:

• Full path to the generated folder

• Generator yielding file paths as they are created

Return type:

tuple[str, Iterator[str]]

Raises:

OSError – If folder creation or file write fails.

ocs_ci.utility.utils.get_acm_mce_build_tag(quay_path, version)

Get Downstream tag from ACM/MCE latest tags

Parameters:

• quay_path (str) – quay image path for acm/mce builds

• version (str) – Version of acm/mce

Returns:

ACM/MCE associated tag

Return type:

str

ocs_ci.utility.utils.get_acm_version(namespace='open-cluster-management')

Get ACM version from CSV

Returns:

returns version string

Return type:

str

ocs_ci.utility.utils.get_architecture_host()

Gets the architecture of host

Returns:

Host architecture

Return type:

str

ocs_ci.utility.utils.get_attr_chain(obj, attr_chain)

Attempt to retrieve object attributes when uncertain about the existence of the attribute or a different attribute in a given attribute chain. If the retrieval fails, None is returned. The function can be used to retrieve a direct attribute, or a chain of attributes. i.e. - obj.attr_a, obj_attr_a.sub_attr

Another example - trying to access “sub_attr_b” in object.attr.sub_attr_a.sub_attr_b - get_attr_chain(object, “attr.sub_attr_a.sub_attr_b”)

The function can be used to try and retrieve “sub_attribute_b” without an exception, even in cases where “attr” or “sub_attr_a” might not exist. In those cases, the function will return None.

Parameters:

• obj – An object

• attr_chain (str) – A string containing one attribute or several sub-attributes separated by dots (i.e. - “attr.sub_attr_a.sub_attr_b”)

Returns:

The requested attribute if found, otherwise None

ocs_ci.utility.utils.get_available_ocp_versions(channel)

Find all available OCP versions for specific channel.

Parameters:

channel (str) – Channel of OCP (e.g. stable-4.2 or fast-4.2)

Returns

list: Sorted list with OCP versions for specified channel.

ocs_ci.utility.utils.get_az_count()

Using a number of different configuration attributes, determine how many availability zones the cluster is configured for.

Returns:

number of availability zones

Return type:

int

ocs_ci.utility.utils.get_ceph_crashes(toolbox_pod)

Gets all Ceph crashes

Parameters:

toolbox_pod (obj) – Ceph toolbox pod object

Returns:

List of ceph crash ID’s

Return type:

list

ocs_ci.utility.utils.get_ceph_version()

Gets the ceph version

Returns:

ceph version

Return type:

str

ocs_ci.utility.utils.get_client_info(ceph_nodes, clients)

ocs_ci.utility.utils.get_client_type_by_name(cluster_name)

Get the client type by the cluster name

Parameters:

cluster_name (str) – The cluster name

Returns:

The client type in lowercase(e.g. kubevirt, agent, non_hosted, etc.,)

Return type:

str

ocs_ci.utility.utils.get_client_version(client_binary_path)

Get version reported by oc version.

Parameters:

client_binary_path (str) – path to oc binary

Returns:

version reported by oc version.

None if the client does not exist at the provided path.

Return type:

str

ocs_ci.utility.utils.get_cluster_id(cluster_path)

Get ClusterID from metadata.json in given cluster_path

Parameters:

cluster_path – path to cluster install directory

Returns:

metadata.json[‘clusterID’]

Return type:

str

ocs_ci.utility.utils.get_cluster_image()

Gets the cluster image

Returns:

cluster image

Return type:

str

ocs_ci.utility.utils.get_cluster_name(cluster_path)

Get clusterName from metadata.json in given cluster_path

Parameters:

cluster_path – path to cluster install directory

Returns:

metadata.json[‘clusterName’]

Return type:

str

ocs_ci.utility.utils.get_cluster_version()

Gets the cluster version

Returns:

cluster version

Return type:

str

ocs_ci.utility.utils.get_cluster_version_info()

Gets the complete cluster version information

Returns:

cluster version information

Return type:

dict

ocs_ci.utility.utils.get_csi_versions()

Gets the CSI related version information

Returns:

CSI related version information

Return type:

dict

ocs_ci.utility.utils.get_default_if_keyval_empty(dictionary, key, default_val)

if Key has an empty value OR key doesn’t exist then return default value

Parameters:

• dictionary (dict) – Dictionary where we have to lookup

• key (str) – key to lookup

• default_val (str) – If key doesn’t have value then return this default_val

Returns:

dictionary[key] if value is present else default_val

ocs_ci.utility.utils.get_glibc_version()

Gets the GLIBC version.

Returns:

GLIBC version

Return type:

str

ocs_ci.utility.utils.get_image_with_digest(image)

Return image with sha256 digest for usage in disconnected environment

Parameters:

image (str) – image

Raises:

UnexpectedImage – In case the image information is unexpected

Returns:

image with sha256 digest specification

Return type:

str

ocs_ci.utility.utils.get_infra_id(cluster_path)

Get infraID from metadata.json in given cluster_path

Parameters:

cluster_path – path to cluster install directory

Returns:

metadata.json[‘infraID’]

Return type:

str

ocs_ci.utility.utils.get_infra_id_from_openshift_install_state(cluster_path)

Get infraID from openshift_install_state.json in given cluster_path

Parameters:

cluster_path – path to cluster install directory

Returns:

cluster infraID

Return type:

str

ocs_ci.utility.utils.get_kubeadmin_password()

ocs_ci.utility.utils.get_latest_acm_tag_unreleased(version)

Get Latest tag for acm unreleased image

Args:

version (str): version of acm for getting latest tag

Returns:

image tag for the specified version

Return type:

str

Raises:

TagNotFoundException – When the given version is not found

ocs_ci.utility.utils.get_latest_ds_olm_tag(upgrade=False, latest_tag=None, stable_upgrade_version=False)

This function returns latest tag of OCS downstream registry or one before latest if upgrade parameter is True

Parameters:

• upgrade (bool) – If True then it returns one version of the build before the latest.

• latest_tag (str) – Tag of the latest build. If not specified config.DEPLOYMENT[‘default_latest_tag’] or ‘latest’ will be used.

• stable_upgrade_version (bool) – If True then it returns stable upgrade version.

Returns:

latest tag for downstream image from quay registry

Return type:

str

Raises:

TagNotFoundException – In case no tag found

ocs_ci.utility.utils.get_latest_ocp_multi_image(version: str = None) → str

Get latest OCP multi-arch image tag for given major.minor version. Optimized with paging early-exit and debug logging.

Parameters:

version (str) – OCP version to get the latest multi-arch image for, if not provided, the current OCP version will be used.

Returns:

Latest OCP multi-arch image:tag

Return type:

str

Raises:

TagNotFoundException – If no matching tag is found

ocs_ci.utility.utils.get_latest_ocp_version(channel, index=-1)

Find latest OCP version for specific channel.

Parameters:

• channel (str) – Channel of OCP (e.g. stable-4.2 or fast-4.2)

• index (int) – Index to get from all available versions list e.g. default -1 is latest version (version[-1]). If you want to get previous version pass index -2 and so on.

Returns

str: Latest OCP version for specified channel.

ocs_ci.utility.utils.get_latest_release_version()

Fetch the latest supported release version of OpenShift from its official mirror site.

Returns:

The latest release version. Example: As of 22 May 2024 the function returns string “4.15.14”

Return type:

str

ocs_ci.utility.utils.get_module_ip(terraform_state_file, module)

Gets the node IP from terraform.tfstate file

Parameters:

• terraform_state_file (str) – Path to terraform state file

• module (str) – Module name in terraform.tfstate file e.g: constants.LOAD_BALANCER_MODULE

Returns:

IP of the node

Return type:

list

ocs_ci.utility.utils.get_nb_db_pod_by_cnpg_label(cnpg_label, namespace='openshift-storage')

Get one of the NooBaa DB CNPG pods

Parameters:

• cnpg_label (str) – The CNPG label that describes the pod role

• namespace (str) – The namespace of the NooBaa DB pod

Returns:

The pod

Return type:

Pod

Raises:

ResourceNotFoundError – If no NooBaa DB pod is found

ocs_ci.utility.utils.get_next_version_available_for_upgrade(current_tag)

This function returns the tag built after the current_version

Parameters:

current_tag (str) – Current build tag from which to search the next one build tag.

Returns:

tag for downstream image from quay registry built after

the current_tag.

Return type:

str

Raises:

TagNotFoundException – In case no tag suitable for upgrade found

ocs_ci.utility.utils.get_nightly_oc_via_ga(version, tarball='openshift-client.tar.gz')

Downloads the nightly OC via GA’ed version

Parameters:

• version (str) – nightly OC version to download

• tarball (str) – target name of the tarfile

ocs_ci.utility.utils.get_nightly_release_info(version)

Get full release information for a nightly build including pullSpec.

Parameters:

version (str) – Nightly version string (e.g., 4.22.0-0.nightly or 4.22.0-0.nightly-2026-04-18-162844)

Returns:

Release information including ‘name’, ‘pullSpec’, ‘downloadURL’, etc.

Returns None if unable to fetch the information.

Return type:

dict

ocs_ci.utility.utils.get_noobaa_cli_config()

Get the appropriate NooBaa CLI path and command prefix based on OCS version.

For OCS version >= 4.20: Returns odf-cli path with ‘noobaa’ prefix For OCS version < 4.20: Returns mcg-cli path with empty prefix

Returns:

(cli_path, command_prefix) where cli_path is the path to CLI binary

and command_prefix is the prefix to use with the command

Return type:

tuple

ocs_ci.utility.utils.get_oadp_version(namespace='openshift-adp')

Returns:

returns version string

Return type:

str

ocs_ci.utility.utils.get_ocm_cli(version=None, bin_dir=None, force_download=False)

Download the OCM binary, if not already present. Update env. PATH and get path of the OCM binary.

Parameters:

• version (str) – Version of the OCM to download

• bin_dir (str) – Path to bin directory (default: config.RUN[‘bin_dir’])

• force_download (bool) – Force OCM download even if already present

Returns:

Path to the OCM binary

Return type:

str

ocs_ci.utility.utils.get_ocp_repo(rhel_major_version=None)

Get ocp repo file, name will be generated dynamically based on ocp version.

Parameters:

rhel_major_version (int) – Major version of RHEL. If not specified it will take major version from config.ENV_DATA[“rhel_version”]

Returns:

Path to ocp repo file

Return type:

string

ocs_ci.utility.utils.get_ocp_upgrade_history()

Gets the OCP upgrade history for the cluster

Returns:

List of OCP upgrade paths. Latest version in the

beginning of the list

Return type:

list

ocs_ci.utility.utils.get_ocp_version(seperator=None)

The deprecated form of ‘get current ocp version’ Use ocs_ci/utility/version.py:get_semantic_ocp_version_from_config()

Get current ocp version

Parameters:

seperator (str) – String that would seperate major and minor version nubers

Returns:

If seperator is ‘None’, version string will be returned as is

eg: ‘4.2’, ‘4.3’. If seperator is provided then ‘.’ in the version string would be replaced by seperator and resulting string will be returned. eg: If seperator is ‘_’ then string returned would be ‘4_2’

Return type:

string

ocs_ci.utility.utils.get_ocs_build_number()

Gets the build number for ocs operator

Returns:

build number for ocs operator version

Return type:

str

ocs_ci.utility.utils.get_ocs_olm_operator_tags(limit=100)

Query the OCS OLM Operator repo and retrieve a list of tags. Since we are limited to 100 tags per page, we end up making several API calls and combining the results into a single list of tags.

Parameters:

limit – the number of tags to limit the request to

Raises:

• KeyError – if the auth config isn’t setup properly

• requests.RequestException – if the response return code is not ok

Returns:

OCS OLM Operator tags

Return type:

list

ocs_ci.utility.utils.get_ocs_version_from_image(image)

Parse major.minor version from OCS image tag.

Parameters:

image (str) – image in format url:tag

Returns

str: Version in x.y format

Raises:

ValueError – In case of the tag which we cannot parse to version.

ocs_ci.utility.utils.get_openshift_client(version=None, bin_dir=None, force_download=False)

Download the OpenShift client binary, if not already present. Update env. PATH and get path of the oc binary.

Parameters:

• version (str) – Version of the client to download (default: config.RUN[‘client_version’])

• bin_dir (str) – Path to bin directory (default: config.RUN[‘bin_dir’])

• force_download (bool) – Force client download even if already present

Returns:

Path to the client binary

Return type:

str

ocs_ci.utility.utils.get_openshift_installer(version=None, bin_dir=None, force_download=False)

Download the OpenShift installer binary, if not already present. Update env. PATH and get path of the openshift installer binary.

Parameters:

• version (str) – Version of the installer to download

• bin_dir (str) – Path to bin directory (default: config.RUN[‘bin_dir’])

• force_download (bool) – Force installer download even if already present

Returns:

Path to the installer binary

Return type:

str

ocs_ci.utility.utils.get_openshift_mirror_url(file_name, version)

Format url to OpenShift mirror (for client and installer download).

Parameters:

• file_name (str) – Name of file

• version (str) – Version of the installer or client to download

Returns:

Url of the desired file (installer or client)

Return type:

str

Raises:

UnsupportedOSType – In case the OS type is not supported

ocs_ci.utility.utils.get_primary_nb_db_pod(namespace='openshift-storage')

Get the NooBaa DB pod that has been assigned the primary role by the CNPG operator.

Returns:

The NooBaa DB pod object

Return type:

Pod

Raises:

ResourceNotFoundError – If no NooBaa DB pod is found

ocs_ci.utility.utils.get_pytest_fixture_value(request, fixture_name)

Get the value of a fixture name from the request

Parameters:

• request (_pytest.fixtures.SubRequest') – The pytest request fixture

• fixture_name – Fixture for which this request is being performed

Returns:

The fixture value

Return type:

Any

ocs_ci.utility.utils.get_random_letters(size=13)

Generates the random string of 3 characters

Parameters:

size (int) – number of letter characters to generate

Returns:

string of random characters of given size

Return type:

str

ocs_ci.utility.utils.get_random_str(size=13)

generates the random string of given size

Parameters:

size (int) – number of random characters to generate

Returns:

string of random characters of given size

Return type:

str

ocs_ci.utility.utils.get_role_arn_from_sub()

Get the RoleARN from the OCS subscription

Returns:

Role ARN used for ODF deployment

Return type:

role_arn (str)

Raises:

• ClusterNotInSTSModeException (Exception) if cluster –

• not in STS mode –

ocs_ci.utility.utils.get_rook_repo(branch='master', to_checkout=None)

Clone and checkout the rook repository to specific branch/commit.

Parameters:

• branch (str) – Branch name to checkout

• to_checkout (str) – Commit id or tag to checkout

ocs_ci.utility.utils.get_rook_version()

Gets the rook version

Returns:

rook version

Return type:

str

ocs_ci.utility.utils.get_root_disk(node)

Fetches the root (boot) disk for node

Parameters:

node (str) – Node name

Returns:

Root disk

Return type:

str

ocs_ci.utility.utils.get_rosa_cli(version=None, bin_dir=None, force_download=False)

Download the ROSA binary, if not already present. Update env. PATH and get path of the ROSA binary.

Parameters:

• version (str) – Version of the ROSA to download

• bin_dir (str) – Path to bin directory (default: config.RUN[‘bin_dir’])

• force_download (bool) – Force ROSA download even if already present

Returns:

Path to the rosa binary

Return type:

str

ocs_ci.utility.utils.get_running_acm_version()

Get the currently deployed ACM version

Returns:

ACM version

Return type:

string

ocs_ci.utility.utils.get_running_cluster_id()

Get cluster UUID Not relying on metadata.json as user sometimes want to run only with kubeconfig for some tests. For this function to work cluster has to be in running state

Returns:

cluster UUID

Return type:

str

ocs_ci.utility.utils.get_running_ocp_version(separator=None, kubeconfig=None)

Get current running ocp version

Parameters:

• separator (str) – String that would separate major and minor version numbers

• kubeconfig (str) – Path to kubeconfig. Optional.

Returns:

If separator is ‘None’, version string will be returned as is

eg: ‘4.2’, ‘4.3’. If separator is provided then ‘.’ in the version string would be replaced by separator and resulting string will be returned. eg: If separator is ‘_’ then string returned would be ‘4_2’

Return type:

string

ocs_ci.utility.utils.get_secondary_nb_db_pod(namespace='openshift-storage')

Get the NooBaa DB pod that has been assigned the secondary role by the CNPG operator.

Returns:

The NooBaa DB pod object

Return type:

Pod

Raises:

ResourceNotFoundError – If no NooBaa DB pod is found

ocs_ci.utility.utils.get_server_version()

Get version reported by oc version.

Returns:

version reported by oc version.

Return type:

str

ocs_ci.utility.utils.get_system_architecture()

Get output from ‘uname -m’ command run on first worker node.

Returns:

Architecture of system

Return type:

str

ocs_ci.utility.utils.get_terraform(version=None, bin_dir=None)

Downloads the terraform binary

Parameters:

• version (str) – Version of the terraform to download

• bin_dir (str) – Path to bin directory (default: config.RUN[‘bin_dir’])

Returns:

Path to the terraform binary

Return type:

str

ocs_ci.utility.utils.get_terraform_ignition_provider(terraform_dir, version=None)

Downloads the terraform ignition provider

Parameters:

• terraform_dir (str) – Path to terraform working directory

• version (str) – Version of the terraform ignition provider to download

ocs_ci.utility.utils.get_testrun_name()

Prepare testrun ID for Polarion (and other reports).

Returns:

String containing testrun name

Return type:

str

ocs_ci.utility.utils.get_trim_mean(values, percentage=20)

Get the trimmed mean of a list of values. Explanation: This function finds the arithmetic mean of given values, ignoring values outside the given limits.

Parameters:

• values (list) – The list of values

• percentage (int) – The percentage to be trimmed

Returns:

Trimmed mean. In case trimmed mean calculation fails,

the regular mean average is returned

Return type:

float

ocs_ci.utility.utils.get_url_content(url, **kwargs)

Return URL content

Parameters:

• url (str) – URL address to return

• kwargs (dict) – additional keyword arguments passed to requests.get(…)

Returns:

Content of URL

Return type:

str

Raises:

AssertionError – When couldn’t load URL

ocs_ci.utility.utils.get_vault_cli(bind_dir=None, force_download=False)

Download vault based on platform basically for CLI purpose. Binary will be directly put into ocs_ci/bin/ directory

Parameters:

• bind_dir (str) – Path to bin directory (default: config.RUN[‘bin_dir’])

• force_download (bool) – Force vault cli download even if already present

ocs_ci.utility.utils.human_to_bytes_ui(size_str)

Convert human readable size to bytes. Use this function when working with UI pages or when format “MiB”, “KiB” with space separation, is used.

Parameters:

size_str (str) – The size to convert (i.e, “1 GiB” is 1048576 bytes) acceptable units are: “EiB”/”Ei”, “PiB”/”Pi” “TiB”/”Ti”, “GiB”/”Gi”, “MiB”/”Mi”, “KiB”/”Ki”, “B”/”Bytes”

Returns:

The converted size in bytes

Return type:

int

ocs_ci.utility.utils.inspect_image(image, authfile_fo, cluster_config=None)

Inspect image

Parameters:

• image (str) – image to inspect

• authfile_fo (NamedTemporaryFile) – pull-secret required for pulling the given image

• cluster_config (MultiClusterConfig) – Holds the context of a cluster

Returns:

json object of the inspected image

Return type:

dict

ocs_ci.utility.utils.is_cluster_running(cluster_path)

ocs_ci.utility.utils.is_cluster_y_version_upgraded()

Checks whether cluster is upgraded or not

Returns:

True if cluster is upgraded from previous versions

Return type:

bool

ocs_ci.utility.utils.is_emoji(char)

ocs_ci.utility.utils.is_ocp_version_gaed(version)

Checks whether given OCP version is GA’ed or not

Parameters:

version (str) – OCP version ( eg: 4.16, 4.15 )

Returns:

True if OCP is GA’ed otherwise None

Return type:

bool

ocs_ci.utility.utils.is_z_stream_upgrade()

Check whether this is a z-stream upgrade scenario

Returns:

True if its a z-stream upgrade else False

Return type:

bool

ocs_ci.utility.utils.kernel_client_io(client, mounting_dir)

ocs_ci.utility.utils.kernel_client_md5(kernel_clients, md5sum_list2)

ocs_ci.utility.utils.kernel_mount(mounting_dir, mon_node_ip, kernel_clients)

ocs_ci.utility.utils.list_insert_at_position(lst, index, element)

Insert an element into the list at a specific index while shifting all the element one setp right to the index

ocs_ci.utility.utils.load_auth_config()

Load the authentication config YAML from /data/auth.yaml

Raises:

FileNotFoundError – if the auth config is not found

Returns:

A dictionary reprensenting the YAML file

Return type:

dict

ocs_ci.utility.utils.load_config_file(config_file)

Loads config file to the ocs-ci config

Parameters:

config_file (str) – Path to yaml config file.

Raises:

FileNotFoundError – In the case the config file not found.

ocs_ci.utility.utils.login_to_mirror_registry(authfile, cluster_config=None)

Login to mirror registry

Parameters:

authfile (str) – authfile (pull-secret) path

ocs_ci.utility.utils.mask_secrets(plaintext, secrets)

Replace secrets in plaintext with asterisks

Parameters:

• plaintext (str or list) – The plaintext to remove the secrets from or list of strings to remove secrets from

• secrets (list) – List of secret strings to replace in the plaintext

Returns:

The censored version of plaintext

Return type:

str

ocs_ci.utility.utils.mds_fail_over(mds_nodes)

ocs_ci.utility.utils.mirror_image(image, cluster_config=None)

Mirror image to mirror image registry.

Parameters:

• image (str) – image to be mirrored, can be defined just with name or with full url, with or without tag or digest

• cluster_config (MultiClusterConfig) – Config object if single cluster, if its multicluster scenario then we will have MultiClusterConfig object

Returns:

the mirrored image link

Return type:

str

ocs_ci.utility.utils.mkdir_pinning(clients, range1, range2, dir_name, pin_val)

ocs_ci.utility.utils.modify_csv(csv, replace_from, replace_to)

Modify the CSV

Parameters:

• csv (str) – The CSV name

• replace_from (str) – The pattern to replace from in the CSV

• replace_to (str) – The pattern to replace to in the CSV

ocs_ci.utility.utils.move_summary_to_top(soup)

Move summary to the top of the eamil report

ocs_ci.utility.utils.mute_mon_netsplit(namespace=None)

Mute MON_NETSPLIT health warning in Ceph cluster. This is useful for arbiter deployments where network splits are expected.

Parameters:

namespace (str) – Namespace of OCS (default: config.ENV_DATA[‘cluster_namespace’])

ocs_ci.utility.utils.ocsci_log_path()

Construct the full path for the log directory.

Returns:

full path for ocs-ci log directory

Return type:

str

ocs_ci.utility.utils.parse_html_for_email(soup)

Parses the html and filters out the unnecessary data/tags/attributes for email reporting

Parameters:

soup (obj) – BeautifulSoup object

ocs_ci.utility.utils.parse_k8s_timestamp(ts: str) → datetime

Parse RFC3339 timestamp like ‘2024-10-30T12:34:56Z’ to a datetime.

Parameters:

ts (str) – Timestamp string in RFC3339 format.

Returns:

Parsed datetime object, or datetime.min if parsing fails.

Return type:

datetime

ocs_ci.utility.utils.parse_pgsql_logs(data)

Parse the pgsql benchmark data from ripsaw and return the data in list format

Parameters:

data (str) – log data from pgsql bench run

Returns:

data digestable by scripts with below format

e.g.:

[ {1: {‘num_clients’: ‘2’,’num_threads’: ‘7’,’latency_avg’: ‘7’, ‘lat_stddev’: ‘0’, ‘tps_incl’: ‘234’, ‘tps_excl’: ‘243’}, {2: {‘num_clients’: ‘2’,’num_threads’: ‘7’,’latency_avg’: ‘7’, ‘lat_stddev’: ‘0’, ‘tps_incl’: ‘234’, ‘tps_excl’: ‘243’}, {3: {‘num_clients’: ‘2’,’num_threads’: ‘7’,’latency_avg’: ‘7’, ‘lat_stddev’: ‘0’, ‘tps_incl’: ‘234’, ‘tps_excl’: ‘243’}, ] where keys{1,2,3} are run-IDs

Return type:

list_data (list)

ocs_ci.utility.utils.pinned_dir_io(clients, mds_fail_over, num_of_files, range1, range2)

ocs_ci.utility.utils.prepare_bin_dir(bin_dir=None)

Prepare bin directory for OpenShift client and installer

Parameters:

bin_dir (str) – Path to bin directory (default: config.RUN[‘bin_dir’])

ocs_ci.utility.utils.prepare_customized_pull_secret(images=None)

Prepare customized pull-secret containing auth section related to given image(s). If image(s) not defined or no related section is found, it will use whole content of pull-secret.

Parameters:

images (str, list) – image (or images) to match with auth section

Returns:

prepared pull-secret

Return type:

NamedTemporaryFile

ocs_ci.utility.utils.query_quay_for_operator_tags(image: str, headers: dict, limit: int, page: int) → list

Query quay tags for the specified image.

Parameters:

• image (str) – Image to query tags for

• headers (dict) – Request headers

• limit (int) – Maximum number of tags to query

• page (int) – Which page of results to return

Raises:

requests.RequestException – If we do not receive an ok response from quay after several retries.

Returns:

list of tags queried from quay

Return type:

list

ocs_ci.utility.utils.read_file_as_str(filepath)

Reads the file content

Parameters:

filepath (str) – File to read

Returns:

File contents in string

Return type:

str

ocs_ci.utility.utils.remove_ceph_crashes(toolbox_pod)

Deletes the Ceph crashes

Parameters:

toolbox_pod (obj) – Ceph toolbox pod object

ocs_ci.utility.utils.remove_keys_from_tf_variable_file(tf_file, keys)

Removes the keys from the tf files and convert to json format

Parameters:

• tf_file (str) – path to tf file

• keys (list) – list of keys to remove

ocs_ci.utility.utils.replace_content_in_file(file, old, new, match_and_replace_line=False)

Replaces contents in file, if old value is not found, it adds new value to the file

Parameters:

• file (str) – Name of the file in which contents will be replaced

• old (str) – Data to search for

• new (str) – Data to replace the old value

• match_and_replace_line (bool) – If True, it will match a line if old pattern is found in the line. The whole line will be replaced with new content. Otherwise it will replace only old string with new string but the rest of the line will be intact. This is the default option.

ocs_ci.utility.utils.restart_mon_pods(mon_ids)

Restart the MON pods.

Parameters:

mon_ids (list) – List of MON IDs

ocs_ci.utility.utils.run_async(command)

Run command locally and return without waiting for completion

Parameters:

command (str) – The command to run.

Returns:

An open descriptor to be used by the calling function.

Example

command = ‘oc delete pvc pvc1’ proc = run_async(command) ret, out, err = proc.async_communicate()

ocs_ci.utility.utils.run_ceph_health_cmd(namespace, detail=False)

Run the ceph health command

Parameters:

• namespace – Namespace of OCS

• detail – If True, it will run the ceph health command with detail option

Raises:

CommandFailed – In case the rook-ceph-tools pod failed to reach the Ready state.

Returns:

The output of the ceph health command

Return type:

str

ocs_ci.utility.utils.run_cmd(cmd, secrets=None, timeout=600, ignore_error=False, threading_lock=None, silent=False, cluster_config=None, output_file=None, **kwargs)

The deprecated form of exec_cmd. Run an arbitrary command locally

Parameters:

• cmd (str) – command to run

• secrets (list) – A list of secrets to be masked with asterisks This kwarg is popped in order to not interfere with subprocess.run(**kwargs)

• timeout (int) – Timeout for the command, defaults to 600 seconds.

• ignore_error (bool) – True if ignore non zero return code and do not raise the exception.

• threading_lock (threading.RLock) – threading.RLock object that is used for handling concurrent oc commands

• silent (bool) – If True will silent errors from the server, default false

• output_file (str) – path where to write stderr from command - apply only when silent mode is True

Raises:

CommandFailed – In case the command execution fails

Returns:

(str) Decoded stdout of command

ocs_ci.utility.utils.run_cmd_interactive(cmd, prompts_answers, timeout=300, string_answer=False, raise_exception=True)

Handle interactive prompts with answers during subctl command

Parameters:

• cmd (str) – Command to be executed

• prompts_answers (dict) – Prompts as keys and answers as values

• timeout (int) – Timeout in seconds, for pexpect to wait for prompt

• string_answer (bool) – string answer

• raise_exception (bool) – raise excption

Raises:

InteractivePromptException – in case something goes wrong

ocs_ci.utility.utils.run_cmd_multicluster(cmd, secrets=None, timeout=600, ignore_error=False, skip_index=None, **kwargs)

Run command on multiple clusters. Useful in multicluster scenarios This is wrapper around exec_cmd

Parameters:

• cmd (str) – command to be run

• secrets (list) – A list of secrets to be masked with asterisks This kwarg is popped in order to not interfere with subprocess.run(**kwargs)

• timeout (int) – Timeout for the command, defaults to 600 seconds.

• ignore_error (bool) – True if ignore non zero return code and do not raise the exception.

• skip_index (list of int) – List of indexes that needs to be skipped from executing the command

Raises:

CommandFailed – In case the command execution fails

Returns:

of CompletedProcess objects as per cluster’s index in config.clusters

i.e. [cluster1_completedprocess, None, cluster2_completedprocess] if command execution skipped on a particular cluster then corresponding entry will have None

Return type:

list

ocs_ci.utility.utils.save_reports()

Save reports of test run to logs directory

ocs_ci.utility.utils.set_aws_region(region=None)

Exports environment variable AWS_REGION

Parameters:

region (str) – AWS region to export

ocs_ci.utility.utils.set_registry_to_managed_state()

In order to be able to deploy from stage we need to change image registry config to Managed state. More described in BZs: https://bugzilla.redhat.com/show_bug.cgi?id=1806593 https://bugzilla.redhat.com/show_bug.cgi?id=1807471#c3 We need to change to managed state as described here: https://github.com/red-hat-storage/ocs-ci/issues/1436 So this is not suppose to be deleted as WA case we really need to do this operation for OCS deployment as was originally done here: https://github.com/red-hat-storage/ocs-ci/pull/1437 Currently it has to be moved here to enable CA certificate to be properly propagated for the stage deployment as mentioned in BZ.

ocs_ci.utility.utils.set_selinux_permissions(workers=None)

Workaround for #1777384 - enable container_use_cephfs on RHEL workers Ticket: RHSTOR-787, see more details in the issue: #1151

Parameters:

workers (list) – List of worker nodes to set selinux permissions

ocs_ci.utility.utils.skip_for_provider_or_client_if_ocs_version(expressions)

This function evaluates the condition for test skip for provider/client clusters based on expression

Parameters:

• expressions (str OR list) – condition for which we need to check,

• eg – A single expression string ‘>=4.2’ OR A list of expressions like [‘<4.3’, ‘>4.2’], [‘<=4.3’, ‘>=4.2’]

Returns:

‘True’ if test needs to be skipped else ‘False’

ocs_ci.utility.utils.skipif_ocp_version(expressions)

This function evaluates the condition for test skip based on expression

Parameters:

• expressions (str OR list) – condition for which we need to check,

• eg – A single expression string ‘>=4.2’ OR A list of expressions like [‘<4.3’, ‘>4.2’], [‘<=4.3’, ‘>=4.2’]

Returns:

‘True’ if test needs to be skipped else ‘False’

ocs_ci.utility.utils.skipif_ocs_version(expressions)

This function evaluates the condition for test skip based on expression

Parameters:

• expressions (str OR list) – condition for which we need to check,

• eg – A single expression string ‘>=4.2’ OR A list of expressions like [‘<4.3’, ‘>4.2’], [‘<=4.3’, ‘>=4.2’]

Returns:

‘True’ if test needs to be skipped else ‘False’

ocs_ci.utility.utils.skipif_ui_not_support(ui_test)

This function evaluates the condition for ui test skip based on ui_test expression

Parameters:

ui_test (str) – condition for which we need to check,

Returns:

‘True’ if test needs to be skipped else ‘False’

ocs_ci.utility.utils.skipif_upgraded_from(version_list)

This function evaluates the condition to skip a test if the cluster is upgraded from a particular OCS version

Parameters:

version_list (list) – List of versions to check

Returns:

True if test needs to be skipped else False

Return type:

(bool)

ocs_ci.utility.utils.string_chunkify(cstring, csize)

Create string chunks of size csize from cstring and yield chunk by chunk

Parameters:

• cstring (str) – Original string which need to be chunkified

• csize (int) – size of each chunk

ocs_ci.utility.utils.sum_of_two_storage_sizes(storage_size1, storage_size2, convert_size=1024)

Calculate the sum of two storage sizes given as strings. Valid units: “Mi”, “Gi”, “Ti”, “MB”, “GB”, “TB”.

Parameters:

• storage_size1 (str) – The first storage size, e.g., “800Mi”, “100Gi”, “2Ti”.

• storage_size2 (str) – The second storage size, e.g., “700Mi”, “500Gi”, “300Gi”.

• convert_size (int) – Set convert by 1000 or 1024. The default value is 1024.

Returns:

The sum of the two storage sizes as a string, e.g., “1500Mi”, “600Gi”, “2300Gi”.

Return type:

str

Raises:

ValueError – If the units of the storage sizes are not match the Valid units

ocs_ci.utility.utils.switch_to_correct_client_type(client_type)

Switch to the correct client type

Parameters:

client_type (str) – The client type(e.g. Kubevirt, Agent, non_hosted, etc.,)

ocs_ci.utility.utils.switch_to_correct_cluster_at_setup(request)

Switch to the correct cluster index at setup, according to the ‘cluster_type’ fixture parameter provided in the test.

Parameters:

request (_pytest.fixtures.SubRequest') – The pytest request fixture

ocs_ci.utility.utils.truncate_large_base64(output: str, max_base64_size: int = 1024) → str

Truncate large base64 blocks in command output to reduce log noise.

Only truncates base64 strings larger than max_base64_size to preserve small base64-encoded secrets that may need to be decoded for debugging.

Parameters:

• output (str) – Command output to process

• max_base64_size (int) – Maximum size for base64 blocks (default 1024 chars)

Returns:

Output with large base64 blocks truncated

Return type:

str

ocs_ci.utility.utils.truncate_long_lines(output: str, max_line_length: int = 500) → str

Truncate individual lines that exceed max_line_length.

Preserves: - First N/2 chars (includes log prefix and key info) - Last 50 chars (for context) - Adds truncation marker in middle

Parameters:

• output (str) – Command output to process.

• max_line_length (int) – Maximum line length before truncation.

Returns:

Output with long lines truncated.

Return type:

str

ocs_ci.utility.utils.update_container_with_mirrored_image(job_pod_dict)

Update Job or Pod configuration dict with mirrored image (required for disconnected installation).

Parameters:

job_pod_dict (dict) – dictionary with Job or Pod configuration

Returns:

for disconnected installation, returns updated Job or Pod dict,

for normal installation return unchanged job_pod_dict

Return type:

dict

ocs_ci.utility.utils.upload_file(server, localpath, remotepath, user=None, password=None, key_file=None, ssh_connection=None)

Upload a file to remote server

Parameters:

• server (str) – Name of the server to upload

• localpath (str) – Local file to upload

• remotepath (str) – Target path on the remote server. filename should be included

• user (str) – User to use for the remote connection

• password (str) – Password to use for the remote connection

• key_file (str) – Key file to use for the remote connection

• ssh_connection (SSHClient) – SSH connection to use for the remote connection

ocs_ci.utility.utils.use_system_available_client_and_kubectl(client_binary_path, kubectl_binary_path)

Use system available client and kubectl binaries.

Parameters:

• client_binary_path (str) – Path to the client binary

• kubectl_binary_path (str) – Path to the kubectl binary

Returns:

True if system available client and kubectl binaries are available and used

Return type:

bool

ocs_ci.utility.utils.validate_dict_values(input_dict: dict) → bool

Validate that all values in the dictionary are the same when ignoring the last two digits.

Parameters:

(dict (input_dict) – {str:int}): The dictionary to validate.

Returns:

True if all values pass the validation, False otherwise.

Return type:

bool

ocs_ci.utility.utils.wait_custom_resource_defenition_available(crd_name, timeout=600)

Wait for the custom resource definition to be available

Parameters:

• crd_name (str) – Name of the custom resource definition

• timeout (int) – Time in seconds to wait for the CRD to be available

Returns:

True if CRD is available, False otherwise

Return type:

bool

ocs_ci.utility.utils.wait_for_ceph_health_not_ok(timeout=300, sleep=10)

Wait until the ceph health is NOT OK

ocs_ci.utility.utils.wait_for_co(operator)

Waits for ClusterOperator to created

Parameters:

operator (str) – Name of the ClusterOperator

ocs_ci.utility.utils.wait_for_machineconfigpool_status(node_type, timeout=1900, skip_tls_verify=False, force_delete_pods=False, cluster_kubeconfig='')

Check for Machineconfigpool status

Parameters:

• node_type (str) – The node type to check machineconfigpool status is updated. e.g: worker, master and all if we want to check for all nodes

• timeout (int) – Time in seconds to wait

• skip_tls_verify (bool) – True if allow skipping TLS verification

• force_delete_pods (bool) – if True delete pods stuck at terminating forcefully

• cluster_kubeconfig (str) – Path to kubeconfig file. Important! Does not work with option force_delete_pods=True

ocs_ci.utility.utils.wipe_all_disk_partitions_for_node(node)

Wipes out partition for all disks which has “nvme” prefix

Parameters:

node (str) – Name of the node (OCP Node)

ocs_ci.utility.utils.wipe_partition(node, disk_path)

Wipes out partition for disk using sgdisk

Parameters:

• node (str) – Name of the node (OCP Node)

• disk_path (str) – Disk to wipe partition

ocs_ci.utility.version module

Module for version related util functions.

ocs_ci.utility.version.compare_versions(expression)

Evaluate version comparison expression

Parameters:

expression (str) – expression string like ‘4.11>=4.2’, supported operators are: >,<,=>,=<,==,!=

Returns:

evaluated comparison expression

Return type:

Boolean

ocs_ci.utility.version.drop_z_version(version_str)

Drops the z (patch) version from a semantic version string.

Parameters:

version_str (str) – Version string in the format x.y.z or x.y

Returns:

Version string in the format x.y

Return type:

str

ocs_ci.utility.version.extract_version_from_csv_name(csv_name)

Extract version string from CSV name.

Parameters:

csv_name (str) – CSV name in the format ‘ocs-operator.v<version>-<suffix>’ or ‘odf-operator.v<version>-<suffix>’ Example: ‘ocs-operator.v4.21.2-rhodf’ or ‘odf-operator.v4.22.0-rhodf’

Returns:

Version string (e.g., ‘4.21.2’)

Return type:

str

Raises:

• IndexError – If CSV name format doesn’t match expected pattern

• ValueError – If version cannot be extracted from CSV name

ocs_ci.utility.version.get_dr_cluster_operator_version(namespace='openshift')

Get DR Cluster Operator Version

Returns:

returns version string

Return type:

str

ocs_ci.utility.version.get_dr_hub_operator_version(namespace='openshift')

Get DR Hub Operator Version

Returns:

returns version string

Return type:

str

ocs_ci.utility.version.get_latest_rosa_ocp_version(version, hosted_cp=False)

Returns latest z-stream version available for ROSA.

Parameters:

• version (str) – OCP version in format x.y

• hosted_cp (bool) – Check versions for ROSA Hosted Control Plane (HCP) clusters.

Returns:

Latest available z-stream version

Return type:

str

ocs_ci.utility.version.get_next_ocp_version_rosa(version)

Get the next available minor version for ROSA.

Parameters:

version (str) – OCP version in format x.y.z

Returns:

Next available version for ROSA

Return type:

str

ocs_ci.utility.version.get_ocp_ga_version(channel)

Retrieve the latest GA version for

Parameters:

channel (str) – the OCP version channel to retrieve GA version for

Returns:

latest GA version for the provided channel.

An empty string is returned if no version exists.

Return type:

str

ocs_ci.utility.version.get_ocp_gitops_operator_version(namespace='openshift')

Get OCP Gitops Operator Version

Returns:

returns version string

Return type:

str

ocs_ci.utility.version.get_ocp_version(seperator=None)

The deprecated form of ‘get current ocp version’ Use ocs_ci/utility/version.py:get_semantic_ocp_version_from_config()

Get current ocp version

Parameters:

seperator (str) – String that would seperate major and minor version nubers

Returns:

If seperator is ‘None’, version string will be returned as is

eg: ‘4.2’, ‘4.3’. If seperator is provided then ‘.’ in the version string would be replaced by seperator and resulting string will be returned. eg: If seperator is ‘_’ then string returned would be ‘4_2’

Return type:

string

ocs_ci.utility.version.get_ocp_versions_rosa(yaml_format=False, hosted_cp=False)

Get the list of available versions for ROSA.

Parameters:

• yaml_format (bool) – Use yaml output from rosa command and parse it as yaml.

• hosted_cp (bool) – List versions for ROSA Hosted Control Plane (HCP) clusters.

Returns:

a list of available versions for ROSA in string format

Return type:

str

ocs_ci.utility.version.get_ocs_version_from_csv(only_major_minor=False, ignore_pre_release=False)

Returns semantic OCS Version from the CSV (ODF if version >= 4.9, OCS otherwise)

Parameters:

• only_major_minor (bool) – If True, only major and minor will be parsed.

• ignore_pre_release (bool) – If True, the pre release version will be ignored

Returns:

Object of semantic version for OCS.

Return type:

semantic_version.base.Version

ocs_ci.utility.version.get_odf_multicluster_orchestrator_version(namespace='open-cluster-management')

Get ODF Multicluster Orchestrator Version

Returns:

returns version string

Return type:

str

ocs_ci.utility.version.get_previous_version(version, count=1)

Fetches the nth previous version

Parameters:

• version (str) – Version ( eg: 4.16, 4.16.0-0.nightly-2024-06-25-194629)

• count (int) – previous version count. if count is 1, it will get 1st previous version. if count is 2, it will get 2nd previous version.

Returns:

Previous version ( returns only major and minor version, eg: 4.15 )

Return type:

str

ocs_ci.utility.version.get_running_odf_client_version()

Get current running ODF Client version

Returns:

ODF Client version

Return type:

string

ocs_ci.utility.version.get_running_odf_version()

Get current running ODF version

Returns:

ODF version

Return type:

string

ocs_ci.utility.version.get_semantic_ocp_running_version(separator=None)

Returning running OCP semantic version from cluster.

Parameters:

separator (str) – String that would separate major and minor version numbers

Returns:

Object of semantic version for OCP.

Return type:

semantic_version.base.Version

ocs_ci.utility.version.get_semantic_ocp_version_from_config()

Returning OCP semantic version from config.

Returns:

Object of semantic version for OCP.

Return type:

semantic_version.base.Version

ocs_ci.utility.version.get_semantic_ocs_version_from_config(cluster_config=None)

Returning OCS semantic version from config.

Returns:

Object of semantic version for OCS. cluster_config (MultiClusterConfig): config related to specific cluster in case of multicluster

Return type:

semantic_version.base.Version

ocs_ci.utility.version.get_semantic_running_odf_version()

Get current running ODF semantic version

Returns:

Object of semantic ODF running version.

Return type:

semantic_version.base.Version

ocs_ci.utility.version.get_semantic_version(version, only_major_minor=False, ignore_pre_release=False, only_major=False)

Returning semantic version from provided version as string.

Parameters:

• version (str) – String version (e.g. 4.6)

• only_major_minor (bool) – If True, only major and minor will be parsed.

• ignore_pre_release (bool) – If True, the pre release version will be ignored

• only_major (bool) – If True, only major will be parsed.

Returns:

Object of semantic version.

Return type:

semantic_version.base.Version

ocs_ci.utility.version.get_submariner_operator_version(namespace='submariner-operator')

Get Submariner Operator Version

Returns:

returns version string

Return type:

str

ocs_ci.utility.version.get_volsync_operator_version(namespace='submariner-operator')

Get VolSync Operator Version

Returns:

returns version string

Return type:

str

ocs_ci.utility.version.ocp_version_available_on_rosa(version)

Check if requested version is available on ROSA for upgrade

Parameters:

version (str) – OCP version in format x.y.z

Returns:

True if version is supported, False otherwise

Return type:

bool

ocs_ci.utility.vsphere module

This module contains the vSphere related methods

class ocs_ci.utility.vsphere.VSPHERE(host, user, password, port=443)

Bases: object

wrapper for vSphere

add_disk(vm, size, disk_type='thin', ssd=False)

Attaches disk to VM

Parameters:

• vm (vim.VirtualMachine) – VM instance

• size (int) – size of disk in GB

• disk_type (str) – disk type

• ssd (bool) – if True, mark disk as SSD

add_disks(vm, disk_sizes, disk_type='thin', ssd=False)

Adds multiple disks to the VM

Parameters:

• vm (vim.VirtualMachine) – VM instance

• disk_sizes (list) – List of the disk sizes in GB

• disk_type (str) – disk type

• ssd (bool) – if True, mark disk as SSD

add_disks_with_same_size(num_disks, vm, size, disk_type='thin', ssd=False)

Adds multiple disks to the VM with the same size

Parameters:

• num_disks (int) – number of disks to add

• vm (vim.VirtualMachine) – VM instance

• size (int) – size of disk in GB

• disk_type (str) – disk type

• ssd (bool) – if True, mark disk as SSD

add_interface_to_compute_vms(network_name='VM Network', adapter_type='vmxnet3')

Add idditional interface to VMs in pool run

Parameters:

• network_name (str) – Network to add the interface, default VM Network

• adapter_type (str) – Type of network adapter, default vmxnet3

add_pci_device(vm, pci_device)

Attaches PCI device to VM

Parameters:

• vm (vim.VirtualMachine) – VM instance

• pci_device (vim.vm.PciPassthroughInfo) – PCI device to add

add_rdm_disk(vm, device_name, disk_mode=None, compatibility_mode=None)

Attaches RDM disk to vm

Parameters:

• vm (vim.VirtualMachine) – VM instance

• device_name (str) – Device name to add to VM. e.g:”/vmfs/devices/disks/naa.600304801b540c0125ef160f3048faba”

• disk_mode (str) – Disk mode. By default it will add ‘independent_persistent’. Available modes are ‘append’, ‘independent_nonpersistent’, ‘independent_persistent’, ‘nonpersistent’, ‘persistent’, and ‘undoable’

• compatibility_mode (str) – Compatabilty mode. Either ‘physicalMode’ or ‘virtualMode’. By default it will add ‘physicalMode’.

available_storage_devices(host, datastore_type='VMFS')

Fetches the available storage devices on Host.

Parameters:

• host (vim.HostSystem) – Host instance

• datastore_type (str) – Type of datastore. Either VMFS or vsan By default, it will take VMFS as datastore type.

Returns:

List of storage devices available for use

Return type:

list

change_vm_network_state(ip, dc, label='Network adapter 1', network='VM Network', timeout=10, connect=True)

Connects or disconnects a VM’s network adapter.

Parameters:

• ip (str) – The IP address of the VM.

• dc (str) – The datacenter name where the VM is located.

• label (str, optional) – The label of the network adapter to disconnect. Defaults to “Network adapter 1”.

• network (str, optional) – The name of the network to disconnect from. Defaults to “VM Network”.

• timeout (int, optional) – The maximum time to wait for the operation to complete. Defaults to 10 seconds.

• connect (bool, optional) – True to connect the network adapter, False to disconnect. Defaults to True.

Returns:

True if the operation was successful, False otherwise.

Return type:

bool

check_folder_exists(name, cluster, dc)

Checks whether folder exists in Templates

Parameters:

• name (str) – Folder name

• cluster (str) – Cluster name

• dc (str) – Datacenter name

Returns:

True if folder exists, False otherwise

Return type:

bool

clone_vm(vm_name, template_name, datacenter_name, resource_pool_name, datastore_name, cluster_name, cpus=4, memory=8, root_disk_size=125829120, network_adapter='VM Network', power_on=True, **kwargs)

Clones the VM from template

Parameters:

• vm_name (str) – Name of the VM to create

• template_name (str) – Template name to clone

• datacenter_name (str) – Name of the Datacenter

• resource_pool_name (str) – Name of the Resource Pool

• datastore_name (str) – Name of the Datastore

• cluster_name (str) – Name of the Cluster in Datacenter

• cpus (int) – Number of CPU’s

• memory (int) – Memory in MB

• root_disk_size (int) – Root Disk size in KB

• network_adapter (str) – Name of the Network Adapter

• power_on (bool) – True to power on the VM after cloning

destroy_folder(name, cluster, dc)

Removes the folder from Templates

Parameters:

• name (str) – Folder name

• cluster (str) – Cluster name

• dc (str) – Datacenter name

destroy_pool(pool, dc, cluster)

Deletes the Resource Pool

Parameters:

• pool (str) – Resource pool name

• dc (str) – Datacenter name

• cluster (str) – Cluster name

destroy_vms(vms, remove_disks=False)

Destroys the VM’s

Parameters:

vms (list) – VM instance list

erase_partition(host, device_path)

Erase the partitions on the disk

Parameters:

• host (vim.HostSystem) – Host instance

• device_path (str) – Device path to erase the partition e.g:”/vmfs/devices/disks/naa.910229801b540c0125ef160f3048faba”

find_datacenter_by_name(datacenter_name)

Fetches the Datacenter

Parameters:

datacenter_name (str) – Name of the Datacenter

Returns:

Datacenter instance

Return type:

vim.Datacenter

find_datastore_by_name(datastore_name, datacenter_name)

Fetches the Datastore

Parameters:

• datastore_name (str) – Name of the Datastore

• datacenter_name (str) – Name of the Datacenter

Returns:

Datastore instance

Return type:

vim.Datastore

find_ip_by_vm(vm, datacenter_name, cluster_name, resource_pool_name)

Fetches the IP for the VM

Parameters:

• vm (str) – Name of VM

• datacenter_name – Name of the Datacenter

• cluster_name – Name of the cluster

• resource_pool_name – Name of the Resource Pool

Returns:

IP of the VM

Return type:

str

find_object_by_name(content, name, obj_type, folder=None, recurse=True)

Finds object by given name

Parameters:

• content (vim.ServiceInstanceContent) – Service Instance Content

• name (str) – Name to search

• obj_type (list) – list of vim.type (e.g: For VM’s, type is vim.VirtualMachine For Hosts, type is vim.HostSystem)

• folder (str) – Folder name

• recurse (bool) – True for recursive search

Returns:

Type of vim instance None: If vim.type doesn’t exists

Return type:

vim.type

find_resource_pool_by_name(resource_pool_name)

Fetches the Resource Pool

Parameters:

resource_pool_name (str) – Name of the Resource Pool

Returns:

Resource Pool instance

Return type:

instance

find_vms_without_ip(name, datacenter_name, cluster_name)

Find all VMs without IP from resource pool

Parameters:

• name (str) – Resource pool name

• datacenter_name (str) – Datacenter name

• cluster_name (str) – vSphere Cluster name

Returns:

VM instances (vim.VirtualMachine)

Return type:

list

get_active_partition(host)

Fetches the active partition disk on Host

Parameters:

host (vim.HostSystem) – Host instance

Returns:

Active partition disk

Return type:

str

get_active_partition_from_mount_info(host)

Gets the active partition from mount info

Parameters:

host (vim.HostSystem) – Host instance

Returns:

Active partition disk

Return type:

str

get_all_objs(content, vimtype, folder=None, recurse=True)

Generate objects of type vimtype

Parameters:

• content (vim.ServiceInstanceContent) – Service Instance Content

• vimtype (vim.type) – Type of vim (e.g: For VM’s, type is vim.VirtualMachine For Hosts, type is vim.HostSystem)

• folder (str) – Folder name

• recurse (bool) – True for recursive search

Returns:

Dictionary of objects and corresponding name

e.g:{

‘vim.Datastore:datastore-12158’: ‘datastore1 (1)’, ‘vim.Datastore:datastore-12157’: ‘datastore1 (2)’ }

Return type:

dict

get_all_vms_in_dc(dc)

Fetches all VMs in Datacenter

Parameters:

dc (str) – Datacenter name

Returns:

List of VMs instance in a Datacenter

Return type:

list

get_all_vms_in_pool(name, dc, cluster)

Gets all VM’s in Resource pool

Parameters:

• name (str) – Resource pool name

• dc (str) – Datacenter name

• cluster (str) – Cluster name

Returns:

VM instances (vim.VirtualMachine)

Return type:

list

Raises:

ResourcePoolNotFound – when Resource pool doesn’t exist

get_cluster(name, dc)

Gets the cluster

Parameters:

• name (str) – Cluster name

• dc (str) – Datacenter name

Returns:

Cluster instance

Return type:

vim.ClusterComputeResource

get_compute_vms_in_pool(name, dc, cluster)

Gets all compute VM’s in Resource pool

Parameters:

• name (str) – Resource pool name

• dc (str) – Datacenter name

• cluster (str) – Cluster name

Returns:

VM instances (vim.VirtualMachine)

Return type:

list

property get_content

Retrieves the content

Returns:

Service Instance Content for Host

Return type:

vim.ServiceInstanceContent

get_controller_for_adding_disk(vm)

Gets the controller for adding disk

Parameters:

vm (vim.VirtualMachine) – VM instance

Returns:

controller instance

get_controllers(vm)

Get the controllers for VM

Parameters:

vm (vim.VirtualMachine) – VM instance

Returns:

list of controllers

Return type:

list

get_datastore_free_capacity(datastore_name, datacenter_name)

Gets the Datastore capacity

Parameters:

• datastore_name (str) – Name of the Datastore

• datacenter_name (str) – Name of the Datacenter

Returns:

Datastore capacity in bytes

Return type:

int

get_datastore_type(datastore)

Gets the Datastore Type

Parameters:

datastore (vim.Datastore) – Datastore instance

Returns:

Datastore type. Either VMFS or vsan

Return type:

str

get_datastore_type_by_name(datastore_name, datacenter_name)

Gets the Datastore Type

Parameters:

• datastore_name (str) – Name of the Datastore

• datacenter_name (str) – Name of the Datacenter

Returns:

Datastore type. Either VMFS or vsan

Return type:

str

get_dc(name)

Gets the Datacenter

Parameters:

name (str) – Datacenter name

Returns:

Datacenter instance

Return type:

vim.Datacenter

get_device_by_key(vm, identifier, key='unit_number')

Get the device by key, and a specific identifier

Parameters:

• vm (vim.VirtualMachine) – VM instance

• identifier (str) – The value of either ‘unit_number’ (Disk unit number to remove), ‘volume_path’ (The volume path in the datastore (i.e, ‘[vsanDatastore] d4210a5e-40ce-efb8-c87e-040973d176e1/control-plane-1.vmdk’), or ‘disk_name’(The disk name (i.e, ‘scsi-36000c290a2cffeb9fcf4a5f748e21909’)

• key (str) – Either ‘unit_number’, ‘volume_path’, or ‘disk_name’

Returns:

The virtual disk device object that

matches the key and the identifier

Return type:

pyVmomi.VmomiSupport.vim.vm.device.VirtualDisk

get_disks(vm)

Fetches the information of all disks in a VM

Parameters:

vm (vim.VirtualMachine) – VM instance

Returns:

List which contains disk related information

Return type:

list

get_host(vm)

Fetches the Host for the VM. Host where VM resides

Parameters:

vm (vim.VirtualMachine) – VM instance

Returns:

Host instance

Return type:

vim.HostSystem

get_host_obj(host_name)

Fetches the Host object

Parameters:

host_name (str) – Host name

Returns:

Host instance

Return type:

vim.HostSystem

get_lunids(dc)

Fetches the LUN ids from the Datacenter

Parameters:

dc (str) – Datacenter name

Returns:

Dictionary contains host name as key and

values as list lun ids

e.g:{

‘HostName1’: [‘02000000193035e73d534’], ‘HostName2’: [‘020000000060034d43333’] }

Return type:

dict

get_mounted_devices(host, datastore_type='VMFS')

Fetches the available storage devices on Host.

Parameters:

• host (vim.HostSystem) – Host instance

• datastore_type (str) – Type of datastore. Either VMFS or vsan By default, it will take VMFS as datastore type.

Returns:

List of storage devices available for use

Return type:

list

get_mounted_devices_in_vmfs(host)

Fetches the devices which was mounted in VMFS

Parameters:

host (vim.HostSystem) – Host instance

Returns:

List of storage devices which was mounted

Return type:

list

get_mounted_devices_in_vsan(host)

Fetches the devices which was mounted in VSAN

Parameters:

host (vim.HostSystem) – Host instance

Returns:

List of storage devices which was mounted

Return type:

list

get_network_device(ip, dc, label='Network adapter 1', network='VM Network')

Get the network adapter for a VM.

Parameters:

• ip (str) – The IP address of the VM.

• dc (str) – The datacenter name where the VM is located.

• label (str, optional) – The label of the network adapter. Defaults to “Network adapter 1”.

• network (str, optional) – The name of the network. Defaults to “VM Network”.

Returns:

The network adapter, or None if not found.

Return type:

vim.vm.device.VirtualEthernetCard

get_passthrough_enabled_devices(vm)

Fetches the passthrough enabled devices

Parameters:

vm (vim.VirtualMachine) – VM instance

Returns:

List of PciPassthroughInfo

Return type:

list

get_pool(name, dc, cluster)

Gets the Resource pool

Parameters:

• name (str) – Resource pool name

• dc (str) – Datacenter name

• cluster (str) – Cluster name

Returns:

Resource pool instance

Return type:

vim.ResourcePool

property get_search_index

Get the search index

Returns:

Instance of Search Index

Return type:

vim.SearchIndex

get_storage_devices(host)

Fetches all the storage devices in the Host. It excludes the enclosures.

Parameters:

host (vim.HostSystem) – Host instance

Returns:

List of storage devices in Host

Return type:

list

get_storage_object_manger()

Gets the vStorageObjectManager

Returns:

vStorageObjectManager

Return type:

vim.vslm.vcenter.VStorageObjectManager

get_unit_number(vm)

Gets the available unit number for the disk

Parameters:

vm (vim.VirtualMachine) – VM instance

Returns:

available unit number for disk

Return type:

int

get_used_devices(host)

Fetches the used storage devices in Host.

Note: Storage devices may be used in different Resource Pools of OCS clusters.

Parameters:

host (vim.HostSystem) – Host instance

Returns:

List of storage devices used

Return type:

list

get_used_unit_number(vm)

Gets the used unit numbers for a VM

Parameters:

vm (vim.VirtualMachine) – VM instance

Returns:

list of unit numbers

Return type:

list

get_used_unit_number_with_all_unit_number(vm)

Gets the used unit numbers including main disk for a VM

Parameters:

vm (vim.VirtualMachine) – VM instance

Returns:

list of unit numbers

Return type:

list

get_vm_by_ip(ip, dc, vm_search=True)

Gets the VM using IP address

Parameters:

• ip (str) – IP address

• dc (str) – Datacenter name

• vm_search (bool) – Search for VMs if True, Hosts if False

Returns:

VM instance

Return type:

vim.VirtualMachine

get_vm_in_pool_by_name(name, dc, cluster, pool)

Gets the VM instance in a resource pool

Parameters:

• name (str) – VM name

• dc (str) – Datacenter name

• cluster (str) – Cluster name

• pool (str) – pool name

Returns:

VM instances

Return type:

vim.VirtualMachine

get_vm_power_status(vm)

Get the VM power status

Parameters:

vm (vm) – VM object

Returns:

VM power status

Return type:

str

get_vms_by_string(str_to_match)

Gets the VM’s with search string

Parameters:

str_to_match (str) – String to match VM’s

Returns:

VM instance

Return type:

list

get_vms_ips(vms)

Get VMs IPs

Parameters:

vms (list) – VM (vm) objects

Returns:

VMs IPs

Return type:

list

get_volume_path(volume_id, datastore_name, datacenter_name)

Gets the Volume path

Parameters:

• volume_id (str) – Volume ID

• datastore_name (str) – Name of the Datastore

• datacenter_name (str) – Name of the Datacenter

Returns:

Path to the volume

Return type:

str

get_vslm_id(volume_id, datastore, storage)

Gets the VSLM ID

Parameters:

• volume_id (str) – Volume ID

• datastore (vim.Datastore) – Datastore instance

• storage (vim.vslm.vcenter.VStorageObjectManager) – vStorageObjectManager

Returns:

vim.vslm.ID

Raises:

VSLMNotFoundException – In case VSLM not found

is_resource_pool_exist(pool, dc, cluster)

Check whether resource pool exists in cluster or not

Parameters:

• pool (str) – Resource pool name

• dc (str) – Datacenter name

• cluster (str) – Cluster name

Returns:

True if resource pool exists, otherwise False

Return type:

bool

is_resource_pool_prefix_exist(pool_prefix, dc, cluster)

Check whether or not resource pool with the provided prefix exist

Parameters:

• pool_prefix (str) – The prefix to look for

• dc (str) – Datacenter name

• cluster (str) – Cluster name

Returns:

True if a resource pool with the same name prefix exists, False otherwise

Return type:

bool

is_template_exist(template_name, dc)

Checks whether template exists in Datacenter

Parameters:

• template_name (str) – template name

• dc (str) – Datacenter name

Returns:

True if template exists, otherwise False

Return type:

bool

is_vm_obj_exist(vm)

Check if the vm object exists.

Parameters:

vm (vim.VirtualMachine) – VM instance

Returns:

True if the VM object exists, False otherwise.

Return type:

bool

map_lunids_to_devices(**kwargs)

Maps the LUN ids to storage devices

Parameters:

**kwargs (dict) – Host to LUN mapping e.g: data = get_lunids(dc) map_lunids_to_devices(**data)

Returns:

Dictionary contains host instance as key and

value as list of device path

Return type:

dict

poweroff_vms(vms)

Powers off the VM and wait for operation to complete

Parameters:

vms (list) – VM instance list

poweron_vms(vms)

Powers on the VM and wait for operation to complete

Parameters:

vms (list) – VM instance list

remove_disk(vm, identifier, key='unit_number', datastore=True)

Removes the Disk from VM and datastore. By default, it will delete the disk ( vmdk ) from VM and backend datastore. If datastore parameter is set to False, then it will ONLY removes the disk from VM

Parameters:

• vm (vim.VirtualMachine) – VM instance

• identifier (str) – The value of either ‘unit_number’ (Disk unit number to remove), ‘volume_path’ (The volume path in the datastore (i.e, ‘[vsanDatastore] d4210a5e-40ce-efb8-c87e-040973d176e1/control-plane-1.vmdk’), or ‘disk_name’(The disk name (i.e, ‘scsi-36000c290a2cffeb9fcf4a5f748e21909’)

• key (str) – Either ‘unit_number’, ‘volume_path’, or ‘disk_name’

• datastore (bool) – Delete the disk (vmdk) from backend datastore if True

remove_disks(vm)

Removes all the extra disks for a VM

Parameters:

vm (vim.VirtualMachine) – VM instance

remove_disks_with_main_disk(vm)

Removes all disks for a VM

Parameters:

vm (vim.VirtualMachine) – VM instance

remove_vms_from_inventory(vms)

Remove the VM’s from inventory

Parameters:

vms (list) – VM instance list

restart_vms(vms, force=False)

Restart VMs by VM Reset or Guest reboot

Parameters:

• vms (list) – VM (vm) objects

• force (bool) – True for Hard reboot(VM Reset), False for Soft reboot(Guest Reboot)

restart_vms_by_stop_and_start(vms, force=True)

Stop and Start VMs

Parameters:

• vms (list) – VM (vm) objects

• force (bool) – True for VM ungraceful power off, False for graceful VM shutdown

start_vms(vms, wait=True)

Start VMs

Parameters:

• vms (list) – VM (vm) objects

• wait (bool) – Wait for VMs to start

stop_vms(vms, force=True, wait=True, timeout=600)

Stop VMs

Parameters:

• vms (list) – VM (vm) objects

• force (bool) – True for VM ungraceful power off, False for graceful VM shutdown

• wait (bool) – Wait for the VMs to stop

• timeout (int) – Timeout in seconds

wait_for_task(task)

Wait for a task to finish

Parameters:

task (instance) – Instance for the task

Returns:

VM instance

Return type:

instance

wait_for_vm_delete(vm, timeout=60)

Wait for the vm object to delete.

Parameters:

• vm (vim.VirtualMachine) – VM instance

• timeout (int) – Time to wait for the VM object to delete.

Returns:

True if the VM object is deleted in the given timeout, False otherwise.

Return type:

bool

wait_for_vm_status(vm, desired_status, timeout=300, interval=10)

Wait for the VM to reach the desired status.

Parameters:

• vm – The virtual machine object

• desired_status – The desired status (e.g., ‘poweredOn’, ‘poweredOff’)

• timeout – Maximum time (in seconds) to wait for the VM to reach the desired status

• interval – Time (in seconds) between each status check

Returns:

True if the VM reaches the desired status, False otherwise

ocs_ci.utility.vsphere_nodes module

Module that contains operations related to vSphere nodes in a cluster This module directly interacts with VM nodes

class ocs_ci.utility.vsphere_nodes.VSPHERENode(host, user=None, private_key=None)

Bases: object

A class that handles operations related to VM node

reboot()

Reboots the node

Returns:

tuple which contains command return code, output and error

Return type:

tuple

replace_ntp_server_in_chrony(server=None)

Replace default NTP server to given server

Parameters:

server (str) – NTP server

restart_chrony()

Restarts chrony service

Returns:

True if successful restarts of chrony, False otherwise

Return type:

bool

restart_service(service_name)

Restarts the given service

set_host_name(host_name)

Sets the host name

Parameters:

host_name (str) – Name to set as host name

Returns:

tuple which contains command return code, output and error

Return type:

tuple

ocs_ci.utility.vsphere_nodes.get_node_ips_from_module(module)

Fetches the node IP’s in cluster from terraform state file

Parameters:

module (str) – Module name in terraform.tfstate file e.g: constants.COMPUTE_MODULE

Returns:

List of module node IP’s

Return type:

list

ocs_ci.utility.vsphere_nodes.update_ntp_and_restart_chrony(node, server=None)

Update the NTP in the node and restarts chronyd service

Parameters:

• node (str) – Hostname/IP of node

• server (str) – NTP server to update in chrony config

ocs_ci.utility.vsphere_nodes.update_ntp_compute_nodes()

Updates NTP server on all compute nodes. In compact mode (worker_replicas == 0), updates master nodes instead.

ocs_ci.utility.workloadfixture module

This module contains functions implementing functionality of workload fixtures in ocs-ci.

ocs_ci.utility.workloadfixture.ignore_next_measurement_file(err, name, test, plugin)

Sets config.RUN[“ignore_next_measurement_file”] to True. This is supposed to be used in monitoring tests that are marked as @flaky to make sure that the measurement is done again.

Reference to the description of flaky rerun_filter function:

https://github.com/box/flaky/blob/d29ca76d5a72845428a0716c74f8fa0d40d98532/flaky/flaky_decorator.py#L16

Parameters:

• err (tuple) – Information about the test failure (from sys.exc_info())

• name (str) – The test name

• test (function) – The test that has raised an error

• plugin (function) – The flaky plugin.

ocs_ci.utility.workloadfixture.is_measurement_done(result_file)

Has the measurement been already performed and stored in a result file?

Returns:

True if the measurement has been already performed.

Return type:

bool

ocs_ci.utility.workloadfixture.measure_operation(operation, result_file, minimal_time=None, metadata=None, measure_after=False, pagerduty_service_ids=None, threading_lock=None)

Get dictionary with keys ‘start’, ‘stop’, ‘metadata’ and ‘result’ that contain information about start and stop time of given function and its result.

Parameters:

• operation (function) – Function to be performed

• result_file (str) – File name that should contain measurement results including logs in json format. If this file exists then it is used for test. Existing file is ignored when config.RUN[“ignore_next_measurement_file”] is set.

• minimal_time (int) – Minimal number of seconds to monitor a system. If provided then monitoring of system continues even when operation is finshed. If not specified then measurement is finished when operation is complete

• metadata (dict) – This can contain dictionary object with information relevant to test (e.g. volume name, operating host, …)

• measure_after (bool) – Determine if time measurement is done before or after the operation returns its state. This can be useful e.g. for capacity utilization testing where operation fills capacity and utilized data are measured after the utilization is completed

• pagerduty_service_ids (list) – Service IDs from PagerDuty system used incidents query

• threading_lock (threading.RLock) – Lock used for synchronization of the threads in Prometheus calls

Returns:

contains information about start and stop time of given

function and its result and provided metadata Example:

{
    'start': 1569827653.1903834,
    'stop': 1569828313.6469617,
    'result': 'rook-ceph-osd-2',
    'metadata': {'status': 'success'},
    'prometheus_alerts': [{'labels': ...}, {...}, ...]
}

Return type:

dict

Module contents