Add unit tests for osism/tasks/openstack.py

osism/tasks/openstack.py

@@ -211,51 +211,23 @@ def get_baremetal_node_ports(node_uuid):
     return port_list
 
 
-def get_baremetal_node_parameters(node_uuid):
-    """Get kernel append params, netplan params, and FRR params for a node.
+def _mask_node_secret_parameters(
+    node_name, kernel_append_params, netplan_parameters, frr_parameters
+):
+    """Mask secret values in a node's kernel/netplan/FRR parameters.
 
-    Extracts parameters from the Ironic node's extra field and masks
-    any secret values (from NetBox secrets custom field) with '***'.
+    Encapsulates the dependency on ``osism.tasks.conductor.utils`` (vault
+    handling and secret masking), so callers -- and their tests -- do not need
+    to reach into conductor implementation details. Secrets are read from the
+    node's NetBox ``secrets`` custom field; any failure along the NetBox/vault
+    path degrades to returning the parameters unmasked.
 
     Returns:
-        dict with kernel_append_params, netplan_parameters, frr_parameters
+        The ``(kernel_append_params, netplan_parameters, frr_parameters)``
+        triple with secret values replaced by ``***``.
     """
-    import json
     from osism.tasks.conductor.utils import deep_decrypt, get_vault, mask_secrets
 
-    conn = utils.get_openstack_connection()
-    node = conn.baremetal.get_node(node_uuid)
-
-    extra = getattr(node, "extra", {}) or {}
-
-    # Parse instance_info from extra (stored as JSON string)
-    instance_info = {}
-    if "instance_info" in extra:
-        try:
-            instance_info = json.loads(extra["instance_info"])
-        except (json.JSONDecodeError, TypeError):
-            pass
-
-    kernel_append_params = instance_info.get("kernel_append_params", "")
-
-    # Parse netplan_parameters from extra
-    netplan_parameters = {}
-    if "netplan_parameters" in extra:
-        try:
-            netplan_parameters = json.loads(extra["netplan_parameters"])
-        except (json.JSONDecodeError, TypeError):
-            pass
-
-    # Parse frr_parameters from extra
-    frr_parameters = {}
-    if "frr_parameters" in extra:
-        try:
-            frr_parameters = json.loads(extra["frr_parameters"])
-        except (json.JSONDecodeError, TypeError):
-            pass
-
-    # Mask secret values in all parameters
-    node_name = getattr(node, "name", None)
     secret_values = set()
     if utils.nb and node_name:
         try:
@@ -326,6 +298,58 @@ def get_baremetal_node_parameters(node_uuid):
             netplan_parameters, mask="***", secret_values=secret_values
         )
 
+    return kernel_append_params, netplan_parameters, frr_parameters
+
+
+def get_baremetal_node_parameters(node_uuid):
+    """Get kernel append params, netplan params, and FRR params for a node.
+
+    Extracts parameters from the Ironic node's extra field and masks
+    any secret values (from NetBox secrets custom field) with '***'.
+
+    Returns:
+        dict with kernel_append_params, netplan_parameters, frr_parameters
+    """
+    import json
+
+    conn = utils.get_openstack_connection()
+    node = conn.baremetal.get_node(node_uuid)
+
+    extra = getattr(node, "extra", {}) or {}
+
+    # Parse instance_info from extra (stored as JSON string)
+    instance_info = {}
+    if "instance_info" in extra:
+        try:
+            instance_info = json.loads(extra["instance_info"])
+        except (json.JSONDecodeError, TypeError):
+            pass
+
+    kernel_append_params = instance_info.get("kernel_append_params", "")
+
+    # Parse netplan_parameters from extra
+    netplan_parameters = {}
+    if "netplan_parameters" in extra:
+        try:
+            netplan_parameters = json.loads(extra["netplan_parameters"])
+        except (json.JSONDecodeError, TypeError):
+            pass
+
+    # Parse frr_parameters from extra
+    frr_parameters = {}
+    if "frr_parameters" in extra:
+        try:
+            frr_parameters = json.loads(extra["frr_parameters"])
+        except (json.JSONDecodeError, TypeError):
+            pass
+
+    node_name = getattr(node, "name", None)
+    kernel_append_params, netplan_parameters, frr_parameters = (
+        _mask_node_secret_parameters(
+            node_name, kernel_append_params, netplan_parameters, frr_parameters
+        )
+    )
+
     return {
         "kernel_append_params": kernel_append_params or None,
         "netplan_parameters": netplan_parameters or None,
@@ -747,6 +771,13 @@ def run_openstack_command_with_cloud(
     )
 
     try:
+        if not cloud_setup_success:
+            logger.error(
+                "Aborting command execution: cloud environment setup failed "
+                f"for cloud '{cloud}'"
+            )
+            return 1
+
         return run_command(
             request_id,
             command,
@@ -783,6 +814,13 @@ def image_manager(
         )
 
         try:
+            if not cloud_setup_success:
+                logger.error(
+                    "Aborting command execution: cloud environment setup failed "
+                    f"for cloud '{cloud}'"
+                )
+                return 1
+
             with tempfile.TemporaryDirectory() as temp_dir:
                 for config in configs:
                     with tempfile.NamedTemporaryFile(
@@ -797,7 +835,8 @@ def image_manager(
                 try:
                     images_index = sanitized_args.index("--images")
                     sanitized_args.pop(images_index)
-                    sanitized_args.pop(images_index)
+                    if images_index < len(sanitized_args):
+                        sanitized_args.pop(images_index)
                 except ValueError:
                     pass
                 sanitized_args.extend(["--images", temp_dir])
@@ -874,6 +913,13 @@ def project_manager(
     )
 
     try:
+        if not cloud_setup_success:
+            logger.error(
+                "Aborting command execution: cloud environment setup failed "
+                f"for cloud '{cloud}'"
+            )
+            return 1
+
         # Change to working directory required by openstack-project-manager
         os.chdir("/openstack-project-manager")
 
@@ -914,6 +960,13 @@ def project_manager_sync(
     )
 
     try:
+        if not cloud_setup_success:
+            logger.error(
+                "Aborting command execution: cloud environment setup failed "
+                f"for cloud '{cloud}'"
+            )
+            return 1
+
         # Change to working directory required by openstack-project-manager
         os.chdir("/openstack-project-manager")
 

tests/unit/tasks/conftest.py

@@ -0,0 +1,22 @@
+# SPDX-License-Identifier: Apache-2.0
+
+"""Fixtures shared across the ``osism.tasks.openstack`` unit test modules.
+
+The suite is split into ``test_openstack_env.py`` (cloud env/connection
+helpers), ``test_openstack_baremetal.py`` (baremetal + NetBox getters and the
+thin Celery task wrappers) and ``test_openstack_managers.py`` (the manager
+tasks). Only fixtures used by more than one of those modules live here;
+module-specific fixtures stay in their module.
+"""
+
+import pytest
+
+
+@pytest.fixture
+def mock_os(mocker):
+    """Replace the module-level ``os`` binding so no test touches the real
+    filesystem or working directory."""
+    fake_os = mocker.patch("osism.tasks.openstack.os")
+    fake_os.getcwd.return_value = "/orig"
+    fake_os.path.exists.return_value = False
+    return fake_os