Add filesystem_freeze module and corresponding unit tests; update ansible-lint exclusions and input-api.yaml

Author: devanshjainms

.ansible-lint

@@ -2,6 +2,7 @@ exclude_paths:
   - src/modules
   - src/module_utils
   - .github
+  - src/roles/ha_db_hana/tasks/files/
 
 skip_list:
   - no-handler

src/module_utils/sap_automation_qa.py

@@ -11,7 +11,6 @@
 import subprocess
 from typing import Optional, Dict, Any
 import xml.etree.ElementTree as ET
-import yaml
 
 
 class TelemetryDataDestination(Enum):
@@ -36,6 +35,9 @@ class TestStatus(Enum):
 
 
 class Parameters:
+    """
+    This class is used to store the parameters for the test case
+    """
 
     def __init__(self, category, id, name, value, expected_value, status):
         self.category = category
@@ -45,7 +47,13 @@ def __init__(self, category, id, name, value, expected_value, status):
         self.expected_value = expected_value
         self.status = status
 
-    def to_dict(self):
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        This method is used to convert the parameters to a dictionary
+
+        :return: Dictionary containing the parameters
+        :rtype: dict
+        """
         return {
             "category": self.category,
             "id": self.id,
@@ -117,9 +125,7 @@ def handle_error(self, exception: Exception, stderr: str = None):
         self.result["message"] = error_message
         self.result["logs"].append(error_message)
 
-    def execute_command_subprocess(
-        self, command: str, shell_command: bool = False
-    ) -> str:
+    def execute_command_subprocess(self, command: str, shell_command: bool = False) -> str:
         """
         Executes a shell command using subprocess with a timeout and logs output or errors.
 
@@ -130,9 +136,7 @@ def execute_command_subprocess(
         :return: Standard output from the command
         :rtype: str
         """
-        command_string = (
-            command if isinstance(command, str) else " ".join(command).replace("'", "")
-        )
+        command_string = command if isinstance(command, str) else " ".join(command).replace("'", "")
         self.log(
             logging.INFO,
             f"Executing command: {command_string}",
@@ -159,7 +163,12 @@ def execute_command_subprocess(
 
     def parse_xml_output(self, xml_output: str) -> Optional[ET.Element]:
         """
-        Parses the XML output of a command.
+        Parses the XML output and returns the root element.
+
+        :param xml_output: XML output to parse
+        :type xml_output: str
+        :return: The root element of the XML output
+        :rtype: Optional[ET.Element]
         """
         if xml_output.startswith("<"):
             return ET.fromstring(xml_output)

src/modules/check_indexserver.py

@@ -5,6 +5,7 @@
 This module is used to check if SAP HANA indexserver is configured.
 """
 
+import logging
 from ansible.module_utils.basic import AnsibleModule
 
 try:
@@ -65,22 +66,28 @@ def check_indexserver(self) -> None:
             )
             return
 
-        global_ini_path = (
-            f"/usr/sap/{self.database_sid}/SYS/global/hdb/custom/config/global.ini"
-        )
-
+        global_ini_path = f"/usr/sap/{self.database_sid}/SYS/global/hdb/custom/config/global.ini"
+        global_ini = []
         try:
             with open(global_ini_path, "r", encoding="utf-8") as file:
                 global_ini = [line.strip() for line in file.readlines()]
 
-            for os_props in (
-                os_props_list if isinstance(os_props_list, list) else [os_props_list]
-            ):
+            self.log(
+                logging.INFO,
+                f"Successfully read the global.ini file: {global_ini}",
+            )
+
+            for os_props in os_props_list if isinstance(os_props_list, list) else [os_props_list]:
                 section_title = list(os_props.keys())[0]
                 if section_title in global_ini:
                     section_start = global_ini.index(section_title)
                     properties_slice = global_ini[section_start + 1 : section_start + 4]
 
+                    self.log(
+                        logging.INFO,
+                        f"Extracted properties: {properties_slice}",
+                    )
+
                     extracted_properties = {
                         prop.split("=")[0].strip(): prop.split("=")[1].strip()
                         for prop in properties_slice

src/modules/filesystem_freeze.py

@@ -4,6 +4,7 @@
 """
 Custom ansible module for formatting the packages list
 """
+import logging
 from typing import Dict, Any
 from ansible.module_utils.basic import AnsibleModule
 
@@ -49,14 +50,18 @@ def run(self) -> Dict[str, Any]:
         """
         file_system = self._find_filesystem()
 
+        self.log(
+            logging.INFO,
+            f"Found the filesystem mounted on /hana/shared: {file_system}",
+        )
+
         if file_system:
-            read_only_output = self.execute_command_subprocess(
-                FREEZE_FILESYSTEM(file_system)
-            )
+            read_only_output = self.execute_command_subprocess(FREEZE_FILESYSTEM(file_system))
+            self.log(logging.INFO, read_only_output)
             self.result.update(
                 {
                     "changed": True,
-                    "message": "The file system (/hana/shared) was successfully mounted as read-only.",
+                    "message": "The file system (/hana/shared) was successfully mounted read-only.",
                     "status": TestStatus.SUCCESS.value,
                     "details": read_only_output,
                 }

src/modules/get_azure_lb.py

@@ -8,9 +8,9 @@
 import logging
 import ast
 from typing import Dict
-from ansible.module_utils.basic import AnsibleModule
 from azure.identity import ManagedIdentityCredential
 from azure.mgmt.network import NetworkManagementClient
+from ansible.module_utils.basic import AnsibleModule
 
 try:
     from ansible.module_utils.sap_automation_qa import (
@@ -55,8 +55,9 @@ def _create_network_client(self):
 
     def get_load_balancers(self) -> list:
         """
-        Get all load balancers in a specific resource group.
+        Get the list of load balancers in a specific resource group.
 
+        :return: List of load balancers
         :rtype: list
         """
         try:
@@ -73,8 +74,9 @@ def get_load_balancers(self) -> list:
 
     def get_load_balancers_details(self) -> dict:
         """
-        Get the details of the DB/SCS/ERS load balancers in a specific resource group.
+        Get the details of the load balancers in a specific resource group.
 
+        :return: Dictionary containing the result of the test case.
         :rtype: dict
         """
         self._create_network_client()
@@ -125,11 +127,13 @@ def check_parameters(entity, parameters_dict, entity_type):
 
         try:
             if found_load_balancer:
+                self.log(
+                    logging.INFO,
+                    f"Found load balancer {found_load_balancer['name']}",
+                )
                 self.result[
                     "message"
-                ] += (
-                    f"Validating load balancer parameters {found_load_balancer['name']}"
-                )
+                ] += f"Validating load balancer parameters {found_load_balancer['name']}"
                 for rule in found_load_balancer["load_balancing_rules"]:
                     try:
                         check_parameters(
@@ -155,16 +159,13 @@ def check_parameters(entity, parameters_dict, entity_type):
                         self.handle_error(e)
                         self.result[
                             "message"
-                        ] += (
-                            f"Failed to validate load balancer probe parameters. {e} \n"
-                        )
+                        ] += f"Failed to validate load balancer probe parameters. {e} \n"
                         continue
 
                 failed_parameters = [
                     param
                     for param in parameters
-                    if param.get("status", TestStatus.ERROR.value)
-                    == TestStatus.ERROR.value
+                    if param.get("status", TestStatus.ERROR.value) == TestStatus.ERROR.value
                 ]
                 self.result.update(
                     {
@@ -176,9 +177,7 @@ def check_parameters(entity, parameters_dict, entity_type):
                         ),
                     }
                 )
-                self.result[
-                    "message"
-                ] += "Successfully validated load balancer parameters"
+                self.result["message"] += "Successfully validated load balancer parameters"
             else:
                 self.result["message"] += "No load balancer found"
 

src/modules/location_constraints.py

@@ -31,9 +31,7 @@ def __init__(self, ansible_os_family: str):
             }
         )
 
-    def remove_location_constraints(
-        self, location_constraints: List[ET.Element]
-    ) -> None:
+    def remove_location_constraints(self, location_constraints: List[ET.Element]) -> None:
         """
         Removes the specified location constraints.
 
@@ -43,8 +41,15 @@ def remove_location_constraints(
         for location_constraint in location_constraints:
             rsc = location_constraint.attrib.get("rsc")
             if rsc:
-                self.execute_command_subprocess(RSC_CLEAR[self.ansible_os_family](rsc))
-                self.result["changed"] = True
+                command_output = self.execute_command_subprocess(
+                    RSC_CLEAR[self.ansible_os_family](rsc)
+                )
+                self.result.update(
+                    {
+                        "details": command_output,
+                        "changed": True,
+                    }
+                )
             else:
                 self.result["changed"] = False
 
@@ -58,11 +63,7 @@ def location_constraints_exists(self) -> List[ET.Element]:
         try:
             xml_output = self.execute_command_subprocess(CONSTRAINTS)
             self.result["details"] = xml_output
-            return (
-                ET.fromstring(xml_output).findall(".//rsc_location")
-                if xml_output
-                else []
-            )
+            return ET.fromstring(xml_output).findall(".//rsc_location") if xml_output else []
         except Exception as e:
             self.handle_exception(e)
 

src/modules/log_parser.py

@@ -98,25 +98,25 @@ def parse_logs(self) -> None:
                             )
                             log_time = log_time.replace(year=start_dt.year)
                         elif self.ansible_os_family == "SUSE":
-                            log_time = datetime.strptime(
-                                line.split(".")[0], "%Y-%m-%dT%H:%M:%S"
-                            )
+                            log_time = datetime.strptime(line.split(".")[0], "%Y-%m-%dT%H:%M:%S")
                         else:
                             continue
 
                         if start_dt <= log_time <= end_dt and any(
                             keyword in line for keyword in self.keywords
                         ):
                             self.result["filtered_logs"].append(
-                                line.translate(
-                                    str.maketrans({"\\": "", '"': "", "'": ""})
-                                )
+                                line.translate(str.maketrans({"\\": "", '"': "", "'": ""}))
                             )
                     except ValueError:
                         continue
 
-            self.result["filtered_logs"] = json.dumps(self.result["filtered_logs"])
-            self.result["status"] = TestStatus.SUCCESS.value
+            self.result.update(
+                {
+                    "filtered_logs": json.dumps(self.result["filtered_logs"]),
+                    "status": TestStatus.SUCCESS.value,
+                }
+            )
         except FileNotFoundError as ex:
             self.handle_error(ex)
         except Exception as e:

src/modules/send_telemetry_data.py

@@ -48,9 +48,7 @@ def __init__(self, module_params: Dict[str, Any]):
         self.result.update(
             {
                 "telemetry_data": module_params["test_group_json_data"],
-                "telemetry_data_destination": module_params[
-                    "telemetry_data_destination"
-                ],
+                "telemetry_data_destination": module_params["telemetry_data_destination"],
                 "start": datetime.now(),
                 "end": datetime.now(),
                 "data_sent": False,
@@ -112,6 +110,10 @@ def send_telemetry_data_to_azuredataexplorer(self, telemetry_json_data: str) ->
         )
         client = QueuedIngestClient(kcsb)
         response = client.ingest_from_dataframe(data_frame, ingestion_properties)
+        self.log(
+            logging.INFO,
+            f"Response from Kusto: {response}",
+        )
         return response
 
     def send_telemetry_data_to_azureloganalytics(
@@ -143,6 +145,10 @@ def send_telemetry_data_to_azureloganalytics(
             },
             timeout=30,
         )
+        self.log(
+            logging.INFO,
+            f"Response from Log Analytics: {response}",
+        )
         return response
 
     def validate_params(self) -> bool:
@@ -151,9 +157,7 @@ def validate_params(self) -> bool:
 
         :return: True if the parameters are valid, False otherwise.
         """
-        telemetry_data_destination = self.module_params.get(
-            "telemetry_data_destination"
-        )
+        telemetry_data_destination = self.module_params.get("telemetry_data_destination")
 
         if telemetry_data_destination == TelemetryDataDestination.LOG_ANALYTICS.value:
             if (
@@ -169,9 +173,7 @@ def validate_params(self) -> bool:
                 "adx_cluster_fqdn",
                 "adx_client_id",
             ]
-            missing_params = [
-                param for param in required_params if param not in self.module_params
-            ]
+            missing_params = [param for param in required_params if param not in self.module_params]
             if missing_params:
                 return False
         return True
@@ -209,9 +211,7 @@ def send_telemetry_data(self) -> None:
             TelemetryDataDestination.KUSTO.value,
             TelemetryDataDestination.LOG_ANALYTICS.value,
         ]:
-            self.log(
-                logging.INFO, "Validating parameters for telemetry data destination "
-            )
+            self.log(logging.INFO, "Validating parameters for telemetry data destination ")
 
             if not self.validate_params():
                 self.result["status"] = (

src/vars/input-api.yaml

@@ -157,3 +157,6 @@ commands:
   - name:                           crm_report_cmd
     SUSE:                           "crm_report -f '{{ test_group_start_time }}' /tmp/{{ test_group_invocation_id }}"
     REDHAT:                         "crm_report -f '{{ test_group_start_time }}' --dest /tmp/{{ test_group_invocation_id }} -yes"
+
+sap_sid:                            "HDB"
+db_sid:                             "HDB"