Add mock data scripts and inventory files for role testing

Author: devanshjainms

tests/roles/ha_db_hana/ha_config_test.py

@@ -0,0 +1,133 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""
+Test class for HANA DB HA config validation tasks
+
+This test class uses pytest to run functional tests on the HANA DB HA config validation tasks
+defined in roles/ha_db/tasks/ha-config.yml. It sets up a temporary test environment,
+mocks necessary Python modules and commands, and verifies the execution of the tasks.
+"""
+
+import os
+import shutil
+from pathlib import Path
+import pytest
+from tests.roles.ha_db_hana.roles_testing_base_db import RolesTestingBaseDB
+
+
+class TestDbHaConfigValidation(RolesTestingBaseDB):
+    """
+    Test class for HANA DB HA config validation tasks.
+    """
+
+    @pytest.fixture
+    def hana_ha_config_tasks(self):
+        """
+        Load the HANA DB HA config validation tasks from the YAML file.
+
+        :return: Parsed YAML content of the tasks file.
+        :rtype: dict
+        """
+        return self.file_operations(
+            operation="read",
+            file_path=Path(__file__).parent.parent.parent
+            / "src/roles/ha_db_hana/tasks/ha-config.yml",
+        )
+
+    @pytest.fixture
+    def test_environment(self, ansible_inventory):
+        """
+        Set up a temporary test environment for the HANA DB HA config validation tasks.
+
+        :param ansible_inventory: Path to the Ansible inventory file.
+        :type ansible_inventory: str
+        :yield temp_dir: Path to the temporary test environment.
+        :ytype: str
+        """
+
+        temp_dir = self.setup_test_environment(
+            role_type="ha_db_hana",
+            ansible_inventory=ansible_inventory,
+            task_name="ha-config",
+            task_description="The HANA DB HA config validation test validates the DB cluster "
+            + "configuration and other system configuration",
+            module_names=[
+                "project/library/get_pcmk_properties_db",
+                "project/library/log_parser",
+                "project/library/send_telemetry_data",
+                "project/library/get_package_list",
+                "bin/crm_resource",
+                "bin/crm",
+            ],
+        )
+
+        os.makedirs(f"{temp_dir}/project/roles/ha_db_hana/tasks/files", exist_ok=True)
+        self.file_operations(
+            operation="write",
+            file_path=f"{temp_dir}/project/roles/ha_db_hana/tasks/files/constants.yaml",
+            content=self.file_operations(
+                operation="read",
+                file_path=Path(__file__).parent.parent / "mock_data/cluster_config.txt",
+            ),
+        )
+
+        os.makedirs(f"{temp_dir}/project/library", exist_ok=True)
+        self.file_operations(
+            operation="write",
+            file_path=f"{temp_dir}/project/library/uri",
+            content=self.file_operations(
+                operation="read",
+                file_path=Path(__file__).parent.parent / "mock_data/azure_metadata.txt",
+            ),
+        )
+        os.chmod(f"{temp_dir}/project/library/uri", 0o755)
+
+        yield temp_dir
+        shutil.rmtree(temp_dir)
+
+    def test_ha_config_validation_success(self, test_environment, ansible_inventory):
+        """
+        Test the HANA DB HA config validation tasks using Ansible Runner.
+
+        :param test_environment: Path to the temporary test environment.
+        :type test_environment: str
+        :param ansible_inventory: Path to the Ansible inventory file.
+        :type ansible_inventory: str
+        """
+        result = self.run_ansible_playbook(
+            test_environment=test_environment, inventory_file_name="inventory_db.txt"
+        )
+
+        assert result.rc == 0, (
+            f"Playbook failed with status: {result.rc}\n"
+            f"STDOUT: {result.stdout.read() if result.stdout else 'No output'}\n"
+            f"STDERR: {result.stderr.read() if result.stderr else 'No errors'}\n"
+            f"Events: {[e.get('event') for e in result.events if 'event' in e]}"
+        )
+
+        ok_events, failed_events = [], []
+        for event in result.events:
+            if event.get("event") == "runner_on_ok":
+                ok_events.append(event)
+            elif event.get("event") == "runner_on_failed":
+                failed_events.append(event)
+
+        assert len(ok_events) > 0
+        assert len(failed_events) == 0
+
+        for event in ok_events:
+            task = event.get("event_data", {}).get("task")
+            task_result = event.get("event_data", {}).get("res")
+            if "Create package dictionary" in task:
+                assert task_result.get("changed") is False
+                assert task_result["details"][1].get("Corosync").get("version") == "2.4.5"
+            if "Virtual Machine name" in task:
+                assert task_result.get("changed") is False
+            if "HA Configuration check" in task:
+                assert task_result.get("changed") is False
+                assert (
+                    task_result["details"].get("parameters", {})[1].get("name")
+                    == "migration-threshold"
+                )
+                assert task_result["details"].get("parameters", {})[1].get("value") == "3"

tests/roles/ha_db_hana/roles_testing_base_db.py

@@ -0,0 +1,44 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""
+Base class for testing roles in Ansible playbooks.
+This class provides a framework for setting up and tearing down test environments,
+mocking necessary modules, and executing Ansible tasks.
+"""
+
+from typing import Iterator
+from pathlib import Path
+import pytest
+from tests.roles.roles_testing_base import RolesTestingBase
+
+
+class RolesTestingBaseDB(RolesTestingBase):
+    """
+    Base class for testing roles in Ansible playbooks.
+    """
+
+    @pytest.fixture
+    def ansible_inventory(self) -> Iterator[str]:
+        """
+        Create a temporary Ansible inventory file for testing.
+        This inventory contains two hosts (db01 and db02) with local connections.
+
+        :yield inventory_path: Path to the temporary inventory file.
+        :ytype: Iterator[str]
+        """
+        inventory_content = self.file_operations(
+            operation="read",
+            file_path=Path(__file__).parent.parent / "mock_data/inventory_db.txt",
+        )
+
+        inventory_path = Path(__file__).parent / "test_inventory.ini"
+        self.file_operations(
+            operation="write",
+            file_path=inventory_path,
+            content=inventory_content,
+        )
+
+        yield str(inventory_path)
+
+        inventory_path.unlink(missing_ok=True)

tests/roles/ha_scs/ascs_migration_test.py

@@ -12,10 +12,10 @@
 import shutil
 from pathlib import Path
 import pytest
-from tests.roles.ha_scs.roles_testing_base import RolesTestingBase
+from tests.roles.ha_scs.roles_testing_base_scs import RolesTestingBaseSCS
 
 
-class TestASCSMigration(RolesTestingBase):
+class TestASCSMigration(RolesTestingBaseSCS):
     """
     Test class for ASCS migration tasks.
     """
@@ -57,6 +57,7 @@ def test_environment(self, ansible_inventory):
         ]
 
         temp_dir = self.setup_test_environment(
+            role_type="ha_scs",
             ansible_inventory=ansible_inventory,
             task_name="ascs-migration",
             task_description="The Resource Migration test validates planned failover scenarios",
@@ -67,7 +68,7 @@ def test_environment(self, ansible_inventory):
                 "bin/crm_resource",
                 "bin/crm",
             ],
-            extra_vars_override={"commands": commands},
+            extra_vars_override={"commands": commands, "node_tier": "scs"},
         )
 
         yield temp_dir
@@ -83,7 +84,7 @@ def test_functional_ascs_migration_success(self, test_environment, ansible_inven
         :type ansible_inventory: str
         """
         result = self.run_ansible_playbook(
-            test_environment=test_environment,
+            test_environment=test_environment, inventory_file_name="inventory_scs.txt"
         )
 
         assert result.rc == 0, (

tests/roles/ha_scs/ascs_node_crash_test.py

@@ -12,10 +12,10 @@
 import shutil
 from pathlib import Path
 import pytest
-from tests.roles.ha_scs.roles_testing_base import RolesTestingBase
+from tests.roles.ha_scs.roles_testing_base_scs import RolesTestingBaseSCS
 
 
-class TestASCSNodeCrash(RolesTestingBase):
+class TestASCSNodeCrash(RolesTestingBaseSCS):
     """
     Test class for ASCS node crash tasks.
     """
@@ -45,6 +45,7 @@ def test_environment(self, ansible_inventory):
         :ytype: str
         """
         temp_dir = self.setup_test_environment(
+            role_type="ha_scs",
             ansible_inventory=ansible_inventory,
             task_name="ascs-node-crash",
             task_description="Simulate ASCS node crash",
@@ -55,6 +56,9 @@ def test_environment(self, ansible_inventory):
                 "bin/crm_resource",
                 "bin/echo",
             ],
+            extra_vars_override={
+                "node_tier": "scs",
+            },
         )
 
         yield temp_dir
@@ -70,7 +74,7 @@ def test_functional_ascs_node_crash_success(self, test_environment, ansible_inve
         :type ansible_inventory: str
         """
         result = self.run_ansible_playbook(
-            test_environment=test_environment,
+            test_environment=test_environment, inventory_file_name="inventory_scs.txt"
         )
 
         assert result.rc == 0, (

tests/roles/ha_scs/ha_config_test.py

@@ -13,10 +13,10 @@
 import shutil
 from pathlib import Path
 import pytest
-from tests.roles.ha_scs.roles_testing_base import RolesTestingBase
+from tests.roles.ha_scs.roles_testing_base_scs import RolesTestingBaseSCS
 
 
-class TestASCSHaConfigValidation(RolesTestingBase):
+class TestASCSHaConfigValidation(RolesTestingBaseSCS):
     """
     Test class for ASCS HA config validation tasks.
     """
@@ -46,6 +46,7 @@ def test_environment(self, ansible_inventory):
         """
 
         temp_dir = self.setup_test_environment(
+            role_type="ha_scs",
             ansible_inventory=ansible_inventory,
             task_name="ha-config",
             task_description="The ASCS HA config validation test validates the SCS/ERS cluster "
@@ -57,6 +58,9 @@ def test_environment(self, ansible_inventory):
                 "bin/crm_resource",
                 "bin/crm",
             ],
+            extra_vars_override={
+                "node_tier": "scs",
+            },
         )
 
         os.makedirs(f"{temp_dir}/project/roles/ha_scs/tasks/files", exist_ok=True)
@@ -65,7 +69,7 @@ def test_environment(self, ansible_inventory):
             file_path=f"{temp_dir}/project/roles/ha_scs/tasks/files/constants.yaml",
             content=self.file_operations(
                 operation="read",
-                file_path=Path(__file__).parent / "mock_data/cluster_config.txt",
+                file_path=Path(__file__).parent.parent / "mock_data/cluster_config.txt",
             ),
         )
 
@@ -75,7 +79,7 @@ def test_environment(self, ansible_inventory):
             file_path=f"{temp_dir}/project/library/uri",
             content=self.file_operations(
                 operation="read",
-                file_path=Path(__file__).parent / "mock_data/azure_metadata.txt",
+                file_path=Path(__file__).parent.parent / "mock_data/azure_metadata.txt",
             ),
         )
         os.chmod(f"{temp_dir}/project/library/uri", 0o755)
@@ -93,7 +97,7 @@ def test_ha_config_validation_success(self, test_environment, ansible_inventory)
         :type ansible_inventory: str
         """
         result = self.run_ansible_playbook(
-            test_environment=test_environment,
+            test_environment=test_environment, inventory_file_name="inventory_scs.txt"
         )
 
         assert result.rc == 0, (

tests/roles/ha_scs/roles_testing_base_scs.py

@@ -0,0 +1,44 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""
+Base class for testing roles in Ansible playbooks.
+This class provides a framework for setting up and tearing down test environments,
+mocking necessary modules, and executing Ansible tasks.
+"""
+
+from typing import Iterator
+from pathlib import Path
+import pytest
+from tests.roles.roles_testing_base import RolesTestingBase
+
+
+class RolesTestingBaseSCS(RolesTestingBase):
+    """
+    Base class for testing roles in Ansible playbooks.
+    """
+
+    @pytest.fixture
+    def ansible_inventory(self) -> Iterator[str]:
+        """
+        Create a temporary Ansible inventory file for testing.
+        This inventory contains two hosts (scs01 and scs02) with local connections.
+
+        :yield inventory_path: Path to the temporary inventory file.
+        :ytype: Iterator[str]
+        """
+        inventory_content = self.file_operations(
+            operation="read",
+            file_path=Path(__file__).parent.parent / "mock_data/inventory_scs.txt",
+        )
+
+        inventory_path = Path(__file__).parent / "test_inventory.ini"
+        self.file_operations(
+            operation="write",
+            file_path=inventory_path,
+            content=inventory_content,
+        )
+
+        yield str(inventory_path)
+
+        inventory_path.unlink(missing_ok=True)