diff --git a/keywords/k8s/k8s_table_parser_base.py b/keywords/k8s/k8s_table_parser_base.py index 6b0f34e2..24bfe770 100644 --- a/keywords/k8s/k8s_table_parser_base.py +++ b/keywords/k8s/k8s_table_parser_base.py @@ -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) diff --git a/keywords/k8s/volumesnapshots/kubectl_get_volumesnapshots_keywords.py b/keywords/k8s/volumesnapshots/kubectl_get_volumesnapshots_keywords.py new file mode 100644 index 00000000..d3421d8a --- /dev/null +++ b/keywords/k8s/volumesnapshots/kubectl_get_volumesnapshots_keywords.py @@ -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.") diff --git a/keywords/k8s/volumesnapshots/object/kubectl_get_volumesnapshots_output.py b/keywords/k8s/volumesnapshots/object/kubectl_get_volumesnapshots_output.py new file mode 100644 index 00000000..97a5dbc7 --- /dev/null +++ b/keywords/k8s/volumesnapshots/object/kubectl_get_volumesnapshots_output.py @@ -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 diff --git a/keywords/k8s/volumesnapshots/object/kubectl_get_volumesnapshots_table_parser.py b/keywords/k8s/volumesnapshots/object/kubectl_get_volumesnapshots_table_parser.py new file mode 100644 index 00000000..e2e99cc8 --- /dev/null +++ b/keywords/k8s/volumesnapshots/object/kubectl_get_volumesnapshots_table_parser.py @@ -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", + ] diff --git a/keywords/k8s/volumesnapshots/object/kubectl_volumesnapshot_object.py b/keywords/k8s/volumesnapshots/object/kubectl_volumesnapshot_object.py new file mode 100644 index 00000000..afc01439 --- /dev/null +++ b/keywords/k8s/volumesnapshots/object/kubectl_volumesnapshot_object.py @@ -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 diff --git a/unit_tests/parser/k8s/kubectl_get_volumesnapshots_table_parser_test.py b/unit_tests/parser/k8s/kubectl_get_volumesnapshots_table_parser_test.py new file mode 100644 index 00000000..f47fb947 --- /dev/null +++ b/unit_tests/parser/k8s/kubectl_get_volumesnapshots_table_parser_test.py @@ -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"