starlingx/test
Code Issues Proposed changes

Merge "Support multiple capabilities in system_storage_backend_object"

Browse Source
This commit is contained in:
Zuul
2025-03-10 12:39:12 +00:00
committed by Gerrit Code Review
parent 9a104d6e8d 391c4d5062
commit f26b40a315
4 changed files with 177 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
keywords/cloud_platform/system/host/objects/storage_capabilities_object.py
View File

@@ -4,43 +4,66 @@ class StorageCapabilities:
"""
def __init__(self):
self.deployment_model = None
self.replication: int = -1
self.min_replication: int = -1
def set_replication(self, replication: int):
"""
Setter for replication
Args:
replication:
Returns:
replication(int): replication number
Returns: None
"""
self.replication = replication
def get_replication(self) -> int:
"""
Getter for replication
Returns:
Returns:
int: replication number
"""
return self.replication
def set_min_replication(self, min_replication: int):
"""
Setter for min_replication
Args:
min_replication:
Returns:
min_replication (int): min_replication number
Returns: None
"""
self.min_replication = min_replication
def get_min_replication(self) -> int:
"""
Getter for min_replication
Returns:
Returns:
int: min_replication number
"""
return self.min_replication
def set_deployment_model(self, deployment_model: str):
"""
Setter for deployment_model
Args:
deployment_model (str): deployment_model
Returns: None
"""
self.deployment_model = deployment_model
def get_deployment_model(self) -> str:
"""
Getter for deployment_model
Returns:
str: deployment_model
"""
return self.deployment_model

110
keywords/cloud_platform/system/storage/objects/system_storage_backend_object.py
View File

@@ -1,9 +1,14 @@
from keywords.cloud_platform.system.host.objects.storage_capabilities_object import StorageCapabilities
import re
from keywords.cloud_platform.system.host.objects.storage_capabilities_object import (
StorageCapabilities,
)
class SystemStorageBackendObject:
"""
This class represents a Storage Backend as an object.
This is typically a line in the system storage-backend-list output table.
"""
@@ -19,137 +24,162 @@ class SystemStorageBackendObject:
def get_name(self) -> str:
"""
Getter for this storage's name
Returns:
str: name
"""
return self.name
def set_uuid(self, uuid: str):
"""
Setter for uuid
Args:
uuid:
Returns:
uuid(str) : uuid
Returns: None
"""
self.uuid = uuid
def get_uuid(self) -> str:
"""
Getter for uuid
Returns:
Returns:
str: uuid
"""
return self.uuid
def set_backend(self, backend: str):
"""
Setter for backend
Args:
backend:
Returns:
backend(str) : backend
Returns: None
"""
self.backend = backend
def get_backend(self) -> str:
"""
Getter for backend
Returns:
Returns:
str: backend
"""
return self.backend
def set_state(self, state: str):
"""
Setter for state
Args:
state:
Returns:
state(str) : state
Returns: None
"""
self.state = state
def get_state(self) -> str:
"""
Getter for state
Returns:
Returns:
str: state
"""
return self.state
def set_task(self, task: str):
"""
Setter for task
Args:
task:
Returns:
task(str) : task
Returns: None
"""
self.task = task
def get_task(self) -> str:
"""
Getter for task
Returns:
Returns:
str: task
"""
return self.task
def set_services(self, services: str):
"""
Setter for services
Args:
services:
Returns:
services(str) : services
Returns: None
"""
self.services = services
def get_services(self) -> str:
"""
Getter for services
Returns:
Returns:
str: service
"""
return self.services
def add_capabilities(self, capabilities_output: str):
"""
Setter for storage capabilities
Adds capabilities to existing object if it is already set.
Setter for storage capabilities or adds capabilities to existing object if it is already set.
Args:
capabilities_output (): the string of capabilities from the system storage-backend-list command
e.g. "replication: 1", "min_replication: 1"
capabilities_output(str): capabilities_output
Raises:
ValueError: If the capabilities output items is not an even number
Raises:
ValueError: If the key name not either replication or min_replication
Returns: None
"""
if not self.capabilities:
self.capabilities = StorageCapabilities()
tokenized_capability = capabilities_output.split(":")
if len(tokenized_capability) != 2:
# str "capabilities_output" is like "replication: 2min_replication: 1"
# Add space after a number if followed by a non-number
capabilities_output = re.sub(r"([0-9]+)([a-zA-Z]+)", r"\1 \2", capabilities_output)
# Add space before "replication" unless preceded by "_" to avoid change for "min_replication"
capabilities_output = re.sub(r"(?<!_)replication", r" replication", capabilities_output)
# Add space before "min_replication"
capabilities_output = re.sub(r"min_", r" min_", capabilities_output)
# Add space before "deployment_model"
capabilities_output = re.sub(r"deploy", r" deploy", capabilities_output)
# remove all colon
capabilities_output = capabilities_output.replace(":", "")
tokenized_capability = capabilities_output.split()
if len(tokenized_capability) % 2 != 0:
raise ValueError(f"Unexpected format for add_capabilities {capabilities_output}. Expecting 'key:value' pair.")
key = tokenized_capability[0].strip()
value = tokenized_capability[1].strip()
if key == 'replication':
self.capabilities.set_replication(int(value))
elif key == 'min_replication':
self.capabilities.set_min_replication(int(value))
else:
raise ValueError("Cannot set value of StorageCapability for key: {key}")
for i in range(0, len(tokenized_capability), 2):
key = tokenized_capability[i]
value = tokenized_capability[i + 1]
if key == "replication":
self.capabilities.set_replication(int(value))
elif key == "min_replication":
self.capabilities.set_min_replication(int(value))
elif key == "deployment_model":
self.capabilities.set_deployment_model(value)
else:
raise ValueError("Cannot set value of StorageCapability for key: {key}")
def get_capabilities(self) -> StorageCapabilities:
"""
Getter for capabilities
Returns:
"""
return self.capabilities

61
keywords/cloud_platform/system/storage/objects/system_storage_backend_output.py
View File

@@ -1,5 +1,7 @@
from framework.exceptions.keyword_exception import KeywordException
from keywords.cloud_platform.system.storage.objects.system_storage_backend_object import SystemStorageBackendObject
from keywords.cloud_platform.system.storage.objects.system_storage_backend_object import (
SystemStorageBackendObject,
)
from keywords.cloud_platform.system.system_table_parser import SystemTableParser
@@ -23,7 +25,6 @@ class SystemStorageBackendOutput:
'+--------------------------------------+------------+---------+------------+------+----------+--------------------+\n']
"""
self.system_storage_backends: [SystemStorageBackendObject] = []
system_table_parser = SystemTableParser(system_storage_backend_list_output)
output_values = system_table_parser.get_output_values_list()
@@ -31,48 +32,64 @@ class SystemStorageBackendOutput:
system_storage_backend_object = None
for value in output_values:
if 'name' not in value:
if "name" not in value:
raise KeywordException(f"The output line {value} was not valid because it is missing an 'name'.")
if not value['name']:
if not value["name"]:
# If there is no name, then this line contains extra info about the last line.
# We should add the capabilities to the last entry.
if 'capabilities' in value:
system_storage_backend_object.add_capabilities(value['capabilities'])
if "capabilities" in value:
system_storage_backend_object.add_capabilities(value["capabilities"])
continue
system_storage_backend_object = SystemStorageBackendObject(value['name'])
system_storage_backend_object = SystemStorageBackendObject(value["name"])
if 'uuid' in value:
system_storage_backend_object.set_uuid(value['uuid'])
if "uuid" in value:
system_storage_backend_object.set_uuid(value["uuid"])
if 'backend' in value:
system_storage_backend_object.set_backend(value['backend'])
if "backend" in value:
system_storage_backend_object.set_backend(value["backend"])
if 'state' in value:
system_storage_backend_object.set_state(value['state'])
if "state" in value:
system_storage_backend_object.set_state(value["state"])
if 'task' in value:
system_storage_backend_object.set_task(value['task'])
if "task" in value:
system_storage_backend_object.set_task(value["task"])
if 'services' in value:
system_storage_backend_object.set_services(value['services'])
if "services" in value:
system_storage_backend_object.set_services(value["services"])
if 'capabilities' in value:
system_storage_backend_object.add_capabilities(value['capabilities'])
if "capabilities" in value:
system_storage_backend_object.add_capabilities(value["capabilities"])
self.system_storage_backends.append(system_storage_backend_object)
def get_system_storage_backends(self) -> list[SystemStorageBackendObject]:
"""
Returns a list of objects representing each row of the table displayed as the result of executing the
'system storage-backend-list' command.
Returns a list of objects representing each row of the table displayed as the result of executing the 'system storage-backend-list' command.
Args: None.
Returns:
list[SystemStorageBackendObject]: list of objects representing each row of the table displayed as the result of executing the
'system storage-backend-list' command.
"""
return self.system_storage_backends
def get_system_storage_backend(self, backend: str) -> SystemStorageBackendObject:
"""
Gets the given backend
Args:
backend (str): the name of the backend ceph or ceph-rook
Raises:
KeywordException: the given name is not exist
Returns:
SystemStorageBackendObject: the given backend object
"""
backends = list(filter(lambda item: item.get_backend() == backend, self.system_storage_backends))
if len(backends) == 0:
raise KeywordException(f"No application with name {backend} was found.")
return backends[0]

37
unit_tests/parser/cloud_platform/storage/storage_backend_table_parser_test.py Normal file
View File

@@ -0,0 +1,37 @@
from keywords.cloud_platform.system.storage.objects.system_storage_backend_output import (
SystemStorageBackendOutput,
)
from keywords.cloud_platform.system.system_table_parser import SystemTableParser
storage_backend_list_ceph_output = ["+--------------------------------------+------------+---------+------------+------+----------+--------------------+\n", "| uuid | name | backend | state | task | services | capabilities |\n", "+--------------------------------------+------------+---------+------------+------+----------+--------------------+\n", "| b3655f85-5e16-471e-b994-c9f910c53f00 | ceph-store | ceph | configured | None | None | replication: 2 |\n", "| | | | | | | min_replication: 1 |\n", "+--------------------------------------+------------+---------+------------+------+----------+--------------------+\n"]
def test_storage_backend_table_parser():
"""
Tests the "system storage_backend_list" table parser
Returns:None
"""
system_table_parser = SystemTableParser(storage_backend_list_ceph_output)
output_values = system_table_parser.get_output_values_list()
assert len(output_values) == 1
output = output_values[0]
assert output["uuid"] == "b3655f85-5e16-471e-b994-c9f910c53f00"
assert output["name"] == "ceph-store"
assert output["backend"] == "ceph"
assert output["state"] == "configured"
assert output["task"] == "None"
assert output["services"] == "None"
assert output["capabilities"] == "replication: 2min_replication: 1"
system_storage_backend_output = SystemStorageBackendOutput(storage_backend_list_ceph_output)
backend_object = system_storage_backend_output.get_system_storage_backend("ceph")
assert backend_object.get_uuid() == "b3655f85-5e16-471e-b994-c9f910c53f00"
assert backend_object.get_name() == "ceph-store"
assert backend_object.get_backend() == "ceph"
assert backend_object.get_state() == "configured"
assert backend_object.get_task() == "None"
assert backend_object.get_services() == "None"
assert backend_object.get_capabilities().get_replication() == 2
assert backend_object.get_capabilities().get_min_replication() == 1