starlingx/test
Code Issues Proposed changes

Add get_volumesnapshots keywords

Browse Source 
- Add get_volumesnapshots keywords
- Fix k8s_table_parser_base.py substring postfix issue
- Add unit test kubectl_get_volumesnapshots_table_parser_test.py

Change-Id: I60784f722da5d87cee7efc86be6b2939a24e0bc2
Signed-off-by: ppeng <peng.peng@windriver.com>
This commit is contained in:
ppeng
2025-10-09 16:07:05 -04:00
parent 89307ac3cf
commit 6e2e722491
6 changed files with 447 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
keywords/k8s/k8s_table_parser_base.py
View File

@@ -1,3 +1,4 @@
import re
from typing import List
from framework.exceptions.keyword_exception import KeywordException
@@ -8,21 +9,25 @@ from keywords.k8s.k8s_table_parser_header import K8sTableParserHeader
class K8sTableParserBase:
"""
Base Class for parsing the output of Table-Like k8s commands.
This class shouldn't be used directly. Instead, it should be inherited by specific k8s table parser implementations.
See KubectlGetPodsTableParser as an example.
"""
def __init__(self, k8s_output):
def __init__(self, k8s_output: str):
"""
Constructor
Args:
k8s_output: The raw String output of a kubernetes command that returns a table.
k8s_output (str): The raw String output of a kubernetes command that returns a table.
"""
self.k8s_output = k8s_output
self.possible_headers = [] # This needs to be defined in child classes of K8sTableParser
def get_output_values_list(self):
"""
get_output_values_list
This function will take the raw String output of a kubernetes command that returns a table
and will parse it into a list of dictionaries. For example, if self.k8s_output is:
@@ -38,7 +43,6 @@ class K8sTableParserBase:
{'NAME': 'default', 'STATUS': 'Active', 'AGE': '18d'}]
"""
if not self.possible_headers:
get_logger().log_error("There are no 'possible_headers' defined. Please use the specific child class of the k8s_table_parser that has the headers that you need.")
raise KeywordException("Undefined 'possible_headers'.")
@@ -71,10 +75,12 @@ class K8sTableParserBase:
def get_headers(self, line: str) -> List[K8sTableParserHeader]:
"""
This function will extract the headers from the header line passed in.
Args:
line: Line containing all the headers to be parsed.
Returns: List of K8sTableParserHeader that have been found, in order.
Args:
line (str): Line containing all the headers to be parsed.
Returns:
List[K8sTableParserHeader]: List of K8sTableParserHeader that have been found, in order.
"""
headers = []
@@ -83,9 +89,10 @@ class K8sTableParserBase:
for header in self.possible_headers:
if header in line:
# Find the header followed by a space or end of line.
# This is to avoid headers that are substrings of other headers.
header_index = line.find(header + " ")
# This is to avoid headers that are substrings of other headers, either substring as prefix or postfix.
pattern = rf"\b{header}\b"
match = re.search(pattern, line)
header_index = match.start()
header_index_last = line.find(header + "\n")
index = max(header_index, header_index_last)

70
keywords/k8s/volumesnapshots/kubectl_get_volumesnapshots_keywords.py Normal file
View File

@@ -0,0 +1,70 @@
import time
from framework.ssh.ssh_connection import SSHConnection
from keywords.base_keyword import BaseKeyword
from keywords.k8s.k8s_command_wrapper import export_k8s_config
from keywords.k8s.volumesnapshots.object.kubectl_get_volumesnapshots_output import KubectlGetVolumesnapshotsOutput
class KubectlGetVolumesnapshotsKeywords(BaseKeyword):
"""
Class for 'kubectl get volumesnapshots.snapshot.storage.k8s.io' keywords
"""
def __init__(self, ssh_connection: SSHConnection):
"""
Initialize the KubectlGetVolumesnapshotsKeywords class.
Args:
ssh_connection (SSHConnection): An SSH connection object to the target system.
"""
self.ssh_connection = ssh_connection
def get_volumesnapshots(self, namespace: str = None, label: str = None) -> KubectlGetVolumesnapshotsOutput:
"""
Gets the k8s volumesnapshots that are available using '-o wide'.
Args:
namespace(str, optional): The namespace to search for volumesnapshots. If None, it will search in all namespaces.
label (str, optional): The label to search for volumesnapshots.
Returns:
KubectlGetVolumesnapshotsOutput: An object containing the parsed output of the command.
"""
arg_namespace = ""
if namespace:
arg_namespace = f"-n {namespace}"
kubectl_get_volumesnapshots_output = self.ssh_connection.send(export_k8s_config(f"kubectl {arg_namespace} -o wide get volumesnapshots.snapshot.storage.k8s.io"))
self.validate_success_return_code(self.ssh_connection)
volumesnapshots_list_output = KubectlGetVolumesnapshotsOutput(kubectl_get_volumesnapshots_output)
return volumesnapshots_list_output
def wait_for_volumesnapshot_status(self, volumesnapshot_name: str, expected_status: str, namespace: str = None, timeout: int = 600) -> bool:
"""
Waits timeout amount of time for the given volumesnapshot to be in the given status
Args:
volumesnapshot_name (str): the volumesnapshot name
expected_status (str): the expected status
namespace (str): the namespace
timeout (int): the timeout in secs
Returns:
bool: True if the volumesnapshot is in the expected status
"""
volumesnapshot_status_timeout = time.time() + timeout
while time.time() < volumesnapshot_status_timeout:
volumesnapshots_output = self.get_volumesnapshots()
if volumesnapshots_output:
volumesnapshot_status = self.get_volumesnapshots(namespace).get_volumesnapshot(volumesnapshot_name).get_ready_to_use()
if volumesnapshot_status == expected_status:
return True
time.sleep(5)
raise ValueError(f"volumesnapshot is not at expected status {expected_status}, after {timeout}s.")

84
keywords/k8s/volumesnapshots/object/kubectl_get_volumesnapshots_output.py Normal file
View File

@@ -0,0 +1,84 @@
from keywords.k8s.volumesnapshots.object.kubectl_get_volumesnapshots_table_parser import KubectlGetVolumesnapshotsTableParser
from keywords.k8s.volumesnapshots.object.kubectl_volumesnapshot_object import KubectlVolumesnapshotObject
class KubectlGetVolumesnapshotsOutput:
"""
A class to interact with and retrieve information about Kubernetes volumesnapshots.snapshot.storage.k8s.io.
This class provides methods to filter and retrieve volumesnapshot information
using the `kubectl` command output.
"""
def __init__(self, kubectl_get_volumesnapshots_output: str):
"""
Constructor
Args:
kubectl_get_volumesnapshots_output (str): Raw string output from running a "kubectl get volumesnapshots" command.
"""
self.kubectl_volumesnapshot: [KubectlVolumesnapshotObject] = []
kubectl_get_volumesnapshots_table_parser = KubectlGetVolumesnapshotsTableParser(kubectl_get_volumesnapshots_output)
output_values_list = kubectl_get_volumesnapshots_table_parser.get_output_values_list()
for volumesnapshot_dict in output_values_list:
if "NAME" not in volumesnapshot_dict:
raise ValueError(f"There is no NAME associated with the volumesnapshot: {volumesnapshot_dict}")
volumesnapshot = KubectlVolumesnapshotObject(volumesnapshot_dict["NAME"])
if "READYTOUSE" in volumesnapshot_dict:
volumesnapshot.set_ready_to_use(volumesnapshot_dict["READYTOUSE"])
if "SOURCEPVC" in volumesnapshot_dict:
volumesnapshot.set_source_pvc(volumesnapshot_dict["SOURCEPVC"])
if "SOURCESNAPSHOTCONTENT" in volumesnapshot_dict:
volumesnapshot.set_source_snapshot_content(volumesnapshot_dict["SOURCESNAPSHOTCONTENT"])
if "RESTORESIZE" in volumesnapshot_dict:
volumesnapshot.set_restore_size(volumesnapshot_dict["RESTORESIZE"])
if "SNAPSHOTCLASS" in volumesnapshot_dict:
volumesnapshot.set_snapshot_class(volumesnapshot_dict["SNAPSHOTCLASS"])
if "SNAPSHOTCONTENT" in volumesnapshot_dict:
volumesnapshot.set_snapshot_content(volumesnapshot_dict["SNAPSHOTCONTENT"])
if "CREATIONTIME" in volumesnapshot_dict:
volumesnapshot.set_creation_time(volumesnapshot_dict["CREATIONTIME"])
if "AGE" in volumesnapshot_dict:
volumesnapshot.set_age(volumesnapshot_dict["AGE"])
self.kubectl_volumesnapshot.append(volumesnapshot)
def get_volumesnapshot(self, volumesnapshot_name: str) -> KubectlVolumesnapshotObject:
"""
This function will get the volumesnapshot with the name specified from this get_volumesnapshots_output.
Args:
volumesnapshot_name (str): The name of the volumesnapshot of interest.
Returns:
KubectlVolumesnapshotObject: The volumesnapshot object with the name specified.
Raises:
ValueError: If the volumesnapshot with the specified name does not exist in the output.
"""
for volumesnapshot in self.kubectl_volumesnapshot:
if volumesnapshot.get_name() == volumesnapshot_name:
return volumesnapshot
else:
raise ValueError(f"There is no volumesnapshot with the name {volumesnapshot_name}.")
def get_volumesnapshots(self) -> [KubectlVolumesnapshotObject]:
"""
Gets all volumesnapshots.
Returns:
[KubectlVolumesnapshotObject]: A list of all volumesnapshots.
"""
return self.kubectl_volumesnapshot

27
keywords/k8s/volumesnapshots/object/kubectl_get_volumesnapshots_table_parser.py Normal file
View File

@@ -0,0 +1,27 @@
from keywords.k8s.k8s_table_parser_base import K8sTableParserBase
class KubectlGetVolumesnapshotsTableParser(K8sTableParserBase):
"""
Class for parsing the output of "kubectl get volumesnapshots.snapshot.storage.k8s.io" commands.
"""
def __init__(self, k8s_output: str):
"""
Constructor
Args:
k8s_output (str): The raw String output of a kubernetes command that returns a table.
"""
super().__init__(k8s_output)
self.possible_headers = [
"NAME",
"READYTOUSE",
"SOURCEPVC",
"SOURCESNAPSHOTCONTENT",
"RESTORESIZE",
"SNAPSHOTCLASS",
"SNAPSHOTCONTENT",
"CREATIONTIME",
"AGE",
]

221
keywords/k8s/volumesnapshots/object/kubectl_volumesnapshot_object.py Normal file
View File

@@ -0,0 +1,221 @@
class KubectlVolumesnapshotObject:
"""
Class to hold attributes of a 'kubectl get volumesnapshots.snapshot.storage.k8s.io' snapshot entry.
"""
def __init__(self, name: str):
"""
Constructor
Args:
name (str): Name of the snapshot.
"""
self.name = name
self.ready_to_use = None
self.source_pvc = None
self.source_snapshot_content = None
self.restore_size = None
self.snapshot_class = None
self.snapshot_content = None
self.creation_time = None
self.age = None
def get_name(self) -> str:
"""
Getter for NAME entry
Returns: The name of the snapshot.
"""
return self.name
def set_ready_to_use(self, ready_to_use: str) -> None:
"""
Setter for READYTOUSE
Args:
ready_to_use (str): 'true' or 'false'
Returns: None
"""
self.ready_to_use = ready_to_use
def get_ready_to_use(self) -> str:
"""
Getter for READYTOUSE entry
"""
return self.ready_to_use
def set_source_pvc(self, source_pvc: str) -> None:
"""
Setter for SOURCEPVC
Args:
source_pvc (str): source_pvc
Returns: None
"""
self.source_pvc = source_pvc
def get_source_pvc(self) -> str:
"""
Getter for SOURCEPVC entry
"""
return self.source_pvc
def set_source_snapshot_content(self, source_snapshot_content: str) -> None:
"""
Setter for SOURCESNAPSHOTCONTENT
Args:
source_snapshot_content (str): source_snapshot_content
Returns: None
"""
self.source_snapshot_content = source_snapshot_content
def get_source_snapshot_content(self) -> str:
"""
Getter for SOURCESNAPSHOTCONTENT entry
"""
return self.source_snapshot_content
def set_restore_size(self, restore_size: str) -> None:
"""
Setter for RESTORESIZE
Args:
restore_size (str): restore_size
Returns: None
"""
self.restore_size = restore_size
def get_restore_size(self) -> str:
"""
Getter for RESTORESIZE entry
"""
return self.restore_size
def set_snapshot_class(self, snapshot_class: str) -> None:
"""
Setter for SNAPSHOTCLASS
Args:
snapshot_class (str): snapshot_class
Returns: None
"""
self.snapshot_class = snapshot_class
def get_snapshot_class(self) -> str:
"""
Getter for SNAPSHOTCLASS entry
"""
return self.snapshot_class
def set_snapshot_content(self, snapshot_content: str) -> None:
"""
Setter for SNAPSHOTCONTENT
Args:
snapshot_content (str): snapshot_content
Returns: None
"""
self.snapshot_content = snapshot_content
def get_snapshot_content(self) -> str:
"""
Getter for SNAPSHOTCONTENT entry
"""
return self.snapshot_content
def set_creation_time(self, creation_time: str) -> None:
"""
Setter for CREATIONTIME
Args:
creation_time (str): creation time
Returns: None
"""
self.creation_time = creation_time
def get_creation_time(self) -> str:
"""
Getter for CREATIONTIME entry
"""
return self.creation_time
def get_creation_time_in_minutes(self) -> int:
"""
Converts the creation_time of the snapshot into minutes.
Returns:
int: The creation_time of the snapshot in minutes.
"""
snapshot_creation_time = self.get_creation_time()
total_minutes = 0
if "m" in snapshot_creation_time:
minutes = int(snapshot_creation_time.split("m")[0])
total_minutes += minutes
if "h" in snapshot_creation_time:
hours = int(snapshot_creation_time.split("h")[0])
total_minutes += hours * 60
if "d" in snapshot_creation_time:
days = int(snapshot_creation_time.split("d")[0])
total_minutes += days * 1440
if "s" in snapshot_creation_time:
pass
return total_minutes
def set_age(self, age: str) -> None:
"""
Setter for AGE
Args:
age (str): ago
Returns: None
"""
self.age = age
def get_age(self) -> str:
"""
Getter for AGE entry
"""
return self.age
def get_age_in_minutes(self) -> int:
"""
Converts the age of the snapshot into minutes.
Returns:
int: The age of the snapshot in minutes.
"""
snapshot_age = self.get_age()
total_minutes = 0
if "m" in snapshot_age:
minutes = int(snapshot_age.split("m")[0])
total_minutes += minutes
if "h" in snapshot_age:
hours = int(snapshot_age.split("h")[0])
total_minutes += hours * 60
if "d" in snapshot_age:
days = int(snapshot_age.split("d")[0])
total_minutes += days * 1440
if "s" in snapshot_age:
pass
return total_minutes

29
unit_tests/parser/k8s/kubectl_get_volumesnapshots_table_parser_test.py Normal file
View File

@@ -0,0 +1,29 @@
from keywords.k8s.volumesnapshots.object.kubectl_get_volumesnapshots_table_parser import KubectlGetVolumesnapshotsTableParser
def test_get_volumesnapshots_table_parser():
"""
Tests the k8s_get_volumesnapshots table parser
Parser k8s_get_volumesnapshots table
"""
get_volumesnapshots_output = (
"NAME READYTOUSE SOURCEPVC SOURCESNAPSHOTCONTENT RESTORESIZE SNAPSHOTCLASS SNAPSHOTCONTENT CREATIONTIME AGE\n",
"mcsi-powerstore-pvc-snapshot true pvol0 8Gi csi-powerstore-snapshot snapcontent-02da0df5-3e9e-4981-a693-f7ae7b03db4c 19m 19m\n",
)
table_parser = KubectlGetVolumesnapshotsTableParser(get_volumesnapshots_output)
output_values = table_parser.get_output_values_list()
assert len(output_values) == 1, "There are two entries in this get volumesnapshots table."
first_line = output_values[0]
assert first_line["NAME"] == "mcsi-powerstore-pvc-snapshot"
assert first_line["READYTOUSE"] == "true"
assert first_line["SOURCEPVC"] == "pvol0"
assert first_line["SOURCESNAPSHOTCONTENT"] == ""
assert first_line["RESTORESIZE"] == "8Gi"
assert first_line["SNAPSHOTCLASS"] == "csi-powerstore-snapshot"
assert first_line["SNAPSHOTCONTENT"] == "snapcontent-02da0df5-3e9e-4981-a693-f7ae7b03db4c"
assert first_line["CREATIONTIME"] == "19m"
assert first_line["AGE"] == "19m"