diff --git a/osism/tasks/openstack.py b/osism/tasks/openstack.py index 54acef9c..5428b296 100644 --- a/osism/tasks/openstack.py +++ b/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, diff --git a/tests/unit/tasks/conftest.py b/tests/unit/tasks/conftest.py new file mode 100644 index 00000000..eaa8cbf3 --- /dev/null +++ b/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 diff --git a/tests/unit/tasks/test_openstack_baremetal.py b/tests/unit/tasks/test_openstack_baremetal.py new file mode 100644 index 00000000..13e2ab94 --- /dev/null +++ b/tests/unit/tasks/test_openstack_baremetal.py @@ -0,0 +1,852 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the baremetal/NetBox getters and the thin Celery task wrappers +of ``osism/tasks/openstack.py``. + +Covers ``get_baremetal_nodes``, the NetBox info getters, ``get_baremetal_node_ports``, +``get_baremetal_node_parameters`` (including its secret-masking helper +``_mask_node_secret_parameters``) and the thin task wrappers around the OpenStack +SDK. + +Every external effect is mocked: the SDK connection comes from a patched +``osism.utils.get_openstack_connection`` (the no-argument helper, via ``mock_conn``) +and the NetBox client replaces the lazy ``osism.utils.nb`` attribute (``mock_nb``). +The bound tasks (``bind=True``) are exercised through ``task.__wrapped__(...)`` +(already bound to the task instance, so ``self.request.id`` is ``None``). + +The masking behavior lives behind ``_mask_node_secret_parameters``, which +encapsulates the dependency on ``osism.tasks.conductor.utils``. Tests of +``get_baremetal_node_parameters`` stub that helper (``stub_mask``), so only the +helper's own tests reach into the conductor internals (``get_vault``, +``deep_decrypt``, ``mask_secrets``, patched at source). +""" + +import json +from operator import attrgetter +from types import SimpleNamespace +from unittest.mock import call + +import pytest + +from osism.tasks import openstack as openstack_tasks + + +def _has_log(records, level, substring): + return any(r["level"] == level and substring in r["message"] for r in records) + + +def _node(extra, name="node-1"): + return SimpleNamespace(extra=extra, name=name) + + +DEFAULT_NETBOX_INFO = { + "device_role": None, + "primary_ip4": None, + "primary_ip6": None, + "netbox_url": None, +} + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def mock_conn(mocker): + """Replace the no-arg ``osism.utils.get_openstack_connection`` helper.""" + conn = mocker.MagicMock() + mocker.patch( + "osism.tasks.openstack.utils.get_openstack_connection", return_value=conn + ) + return conn + + +@pytest.fixture +def mock_nb(mocker): + """Replace ``osism.utils.nb`` (lazy attribute) with a fresh MagicMock.""" + nb = mocker.MagicMock() + mocker.patch("osism.tasks.openstack.utils.nb", new=nb, create=True) + return nb + + +@pytest.fixture +def no_netbox(mocker): + """Force ``osism.utils.nb`` to ``None`` (no NetBox integration).""" + mocker.patch("osism.tasks.openstack.utils.nb", new=None, create=True) + + +@pytest.fixture +def stub_mask(mocker): + """Stub ``_mask_node_secret_parameters`` so ``get_baremetal_node_parameters`` + tests stay decoupled from ``conductor.utils``; by default it returns the + parameters unchanged.""" + return mocker.patch( + "osism.tasks.openstack._mask_node_secret_parameters", + side_effect=lambda node_name, kernel, netplan, frr: (kernel, netplan, frr), + ) + + +@pytest.fixture +def node_params_env(mocker, mock_nb): + """Wire up the NetBox device and the vault helpers for the masking path. + + ``get_vault``/``deep_decrypt`` are imported inside the function under + test, so they are patched at source in ``osism.tasks.conductor.utils``; + ``mask_secrets`` stays real unless a test patches it explicitly. + """ + device = SimpleNamespace(custom_fields={"secrets": {}}) + mock_nb.dcim.devices.get.return_value = device + get_vault = mocker.patch( + "osism.tasks.conductor.utils.get_vault", return_value=mocker.MagicMock() + ) + deep_decrypt = mocker.patch("osism.tasks.conductor.utils.deep_decrypt") + return SimpleNamespace( + nb=mock_nb, + device=device, + get_vault=get_vault, + deep_decrypt=deep_decrypt, + ) + + +# --------------------------------------------------------------------------- +# get_baremetal_nodes +# --------------------------------------------------------------------------- + + +def test_get_baremetal_nodes_maps_all_fields(mock_conn): + node = SimpleNamespace( + uuid="u1", + name="n1", + power_state="power on", + provision_state="active", + maintenance=False, + maintenance_reason="mr", + fault="f", + instance_uuid="i1", + driver="redfish", + resource_class="bm", + conductor="cond", + owner="own", + lessee="les", + description="desc", + allocation_uuid="alloc", + traits=["CUSTOM_X"], + properties={"cpus": 4}, + extra={"k": "v"}, + driver_info={"redfish_address": "https://bmc.example"}, + last_error="err", + provision_updated_at="p-ts", + created_at="c-ts", + updated_at="u-ts", + ) + mock_conn.baremetal.nodes.return_value = iter([node]) + + result = openstack_tasks.get_baremetal_nodes() + + mock_conn.baremetal.nodes.assert_called_once_with(details=True) + assert result == [ + { + "uuid": "u1", + "name": "n1", + "device_role": None, + "primary_ip4": None, + "primary_ip6": None, + "power_state": "power on", + "provision_state": "active", + "maintenance": False, + "maintenance_reason": "mr", + "fault": "f", + "instance_uuid": "i1", + "driver": "redfish", + "resource_class": "bm", + "conductor": "cond", + "owner": "own", + "lessee": "les", + "description": "desc", + "allocation_uuid": "alloc", + "traits": ["CUSTOM_X"], + "properties": {"cpus": 4}, + "extra": {"k": "v"}, + "redfish_address": "https://bmc.example", + "last_error": "err", + "provision_updated_at": "p-ts", + "created_at": "c-ts", + "updated_at": "u-ts", + } + ] + + +def test_get_baremetal_nodes_fallbacks(mock_conn): + """Missing ``uuid`` falls back to ``id``; ``traits``/``driver_info`` of + ``None`` are normalized. ``SimpleNamespace`` models missing attributes.""" + node = SimpleNamespace(id="fallback-id", traits=None, driver_info=None) + mock_conn.baremetal.nodes.return_value = iter([node]) + + (info,) = openstack_tasks.get_baremetal_nodes() + + assert info["uuid"] == "fallback-id" + assert info["name"] is None + assert info["traits"] == [] + assert info["redfish_address"] is None + assert info["properties"] == {} + assert info["extra"] == {} + + +def test_get_baremetal_nodes_empty(mock_conn): + mock_conn.baremetal.nodes.return_value = iter([]) + + assert openstack_tasks.get_baremetal_nodes() == [] + + +# --------------------------------------------------------------------------- +# get_baremetal_node_netbox_info / get_baremetal_nodes_netbox_info +# --------------------------------------------------------------------------- + + +def test_netbox_info_without_netbox_client(no_netbox): + result = openstack_tasks.get_baremetal_node_netbox_info("node-1") + + assert result == DEFAULT_NETBOX_INFO + + +def test_netbox_info_without_node_name(mock_nb): + result = openstack_tasks.get_baremetal_node_netbox_info("") + + assert result == DEFAULT_NETBOX_INFO + mock_nb.dcim.devices.get.assert_not_called() + + +def test_netbox_info_device_found(mocker, mock_nb): + """Role name, split primary IPs and the device URL are extracted.""" + mocker.patch( + "osism.tasks.openstack.settings.NETBOX_URL", "https://netbox.example.com/" + ) + device = SimpleNamespace( + id=42, + role=SimpleNamespace(name="server"), + primary_ip4="10.0.0.1/24", + primary_ip6="fd00::1/64", + ) + mock_nb.dcim.devices.get.return_value = device + + result = openstack_tasks.get_baremetal_node_netbox_info("node-1") + + mock_nb.dcim.devices.get.assert_called_once_with(name="node-1") + assert result == { + "device_role": "server", + "primary_ip4": "10.0.0.1", + "primary_ip6": "fd00::1", + "netbox_url": "https://netbox.example.com/dcim/devices/42/", + } + + +def test_netbox_info_inventory_hostname_fallback(mocker, mock_nb): + """No direct name match: the first ``cf_inventory_hostname`` filter hit is + used. With ``NETBOX_URL`` unset, ``netbox_url`` stays ``None``.""" + mocker.patch("osism.tasks.openstack.settings.NETBOX_URL", None) + device = SimpleNamespace( + id=7, + role=SimpleNamespace(name="switch"), + primary_ip4=None, + primary_ip6=None, + ) + mock_nb.dcim.devices.get.return_value = None + mock_nb.dcim.devices.filter.return_value = [device] + + result = openstack_tasks.get_baremetal_node_netbox_info("node-1") + + mock_nb.dcim.devices.filter.assert_called_once_with(cf_inventory_hostname="node-1") + assert result["device_role"] == "switch" + assert result["netbox_url"] is None + + +def test_netbox_info_no_device_found(mock_nb): + mock_nb.dcim.devices.get.return_value = None + mock_nb.dcim.devices.filter.return_value = [] + + result = openstack_tasks.get_baremetal_node_netbox_info("node-1") + + assert result == DEFAULT_NETBOX_INFO + + +@pytest.mark.parametrize("role", [None, object()], ids=["none", "no-name"]) +def test_netbox_info_role_without_name(mocker, mock_nb, role): + """A missing role or one without a ``name`` leaves ``device_role`` unset.""" + mocker.patch("osism.tasks.openstack.settings.NETBOX_URL", None) + device = SimpleNamespace(id=1, role=role, primary_ip4=None, primary_ip6=None) + mock_nb.dcim.devices.get.return_value = device + + result = openstack_tasks.get_baremetal_node_netbox_info("node-1") + + assert result["device_role"] is None + + +def test_netbox_info_lookup_error_returns_defaults(mock_nb, loguru_logs): + mock_nb.dcim.devices.get.side_effect = Exception("netbox down") + + result = openstack_tasks.get_baremetal_node_netbox_info("node-1") + + assert result == DEFAULT_NETBOX_INFO + assert _has_log(loguru_logs, "DEBUG", "Could not get NetBox info for node-1") + + +def test_nodes_netbox_info_without_netbox_client(no_netbox): + assert openstack_tasks.get_baremetal_nodes_netbox_info(["a"]) == {} + + +def test_nodes_netbox_info_empty_names(mocker, mock_nb): + single = mocker.patch("osism.tasks.openstack.get_baremetal_node_netbox_info") + + assert openstack_tasks.get_baremetal_nodes_netbox_info([]) == {} + + single.assert_not_called() + + +def test_nodes_netbox_info_queries_each_name(mocker, mock_nb): + single = mocker.patch( + "osism.tasks.openstack.get_baremetal_node_netbox_info", + side_effect=lambda name: {"device_role": f"role-{name}"}, + ) + + result = openstack_tasks.get_baremetal_nodes_netbox_info(["a", "b"]) + + assert result == { + "a": {"device_role": "role-a"}, + "b": {"device_role": "role-b"}, + } + assert single.call_args_list == [call("a"), call("b")] + + +# --------------------------------------------------------------------------- +# get_baremetal_node_ports +# --------------------------------------------------------------------------- + + +def test_get_baremetal_node_ports(mock_conn): + port1 = SimpleNamespace( + uuid="p1", + address="aa:bb:cc:dd:ee:ff", + node_uuid="n1", + pxe_enabled=True, + created_at="c-ts", + updated_at="u-ts", + ) + port2 = SimpleNamespace(id="p2") + mock_conn.baremetal.ports.return_value = iter([port1, port2]) + + result = openstack_tasks.get_baremetal_node_ports("n1") + + mock_conn.baremetal.ports.assert_called_once_with(details=True, node_uuid="n1") + assert result[0] == { + "uuid": "p1", + "address": "aa:bb:cc:dd:ee:ff", + "node_uuid": "n1", + "pxe_enabled": True, + "created_at": "c-ts", + "updated_at": "u-ts", + } + assert result[1]["uuid"] == "p2" + assert result[1]["address"] is None + + +def test_get_baremetal_node_ports_empty(mock_conn): + mock_conn.baremetal.ports.return_value = iter([]) + + assert openstack_tasks.get_baremetal_node_ports("n1") == [] + + +# --------------------------------------------------------------------------- +# get_baremetal_node_parameters (orchestration) +# --------------------------------------------------------------------------- +# +# These tests stub ``_mask_node_secret_parameters`` so they exercise only the +# extraction/parsing of the Ironic node's ``extra`` field and the shaping of +# the result dict; the masking helper is tested separately below. + + +@pytest.mark.parametrize("extra", [None, {}], ids=["none", "empty"]) +def test_node_parameters_empty_extra(mock_conn, stub_mask, extra): + mock_conn.baremetal.get_node.return_value = _node(extra) + + result = openstack_tasks.get_baremetal_node_parameters("u1") + + mock_conn.baremetal.get_node.assert_called_once_with("u1") + assert result == { + "kernel_append_params": None, + "netplan_parameters": None, + "frr_parameters": None, + } + + +def test_node_parameters_extracts_kernel_append_params(mock_conn, stub_mask): + extra = {"instance_info": json.dumps({"kernel_append_params": "console=tty0"})} + mock_conn.baremetal.get_node.return_value = _node(extra) + + result = openstack_tasks.get_baremetal_node_parameters("u1") + + assert result["kernel_append_params"] == "console=tty0" + + +@pytest.mark.parametrize("instance_info", ["{invalid", 12345], ids=["json", "type"]) +def test_node_parameters_invalid_instance_info(mock_conn, stub_mask, instance_info): + """Unparsable ``instance_info`` (bad JSON or non-string) is ignored.""" + mock_conn.baremetal.get_node.return_value = _node({"instance_info": instance_info}) + + result = openstack_tasks.get_baremetal_node_parameters("u1") + + assert result["kernel_append_params"] is None + + +def test_node_parameters_parses_netplan_and_frr(mock_conn, stub_mask): + extra = { + "netplan_parameters": json.dumps({"eth0": "dhcp"}), + "frr_parameters": json.dumps({"asn": "65000"}), + } + mock_conn.baremetal.get_node.return_value = _node(extra) + + result = openstack_tasks.get_baremetal_node_parameters("u1") + + assert result["netplan_parameters"] == {"eth0": "dhcp"} + assert result["frr_parameters"] == {"asn": "65000"} + + +def test_node_parameters_invalid_netplan_and_frr(mock_conn, stub_mask): + extra = {"netplan_parameters": "{invalid", "frr_parameters": "{invalid"} + mock_conn.baremetal.get_node.return_value = _node(extra) + + result = openstack_tasks.get_baremetal_node_parameters("u1") + + assert result["netplan_parameters"] is None + assert result["frr_parameters"] is None + + +def test_node_parameters_forwards_parsed_values_to_masking(mock_conn, stub_mask): + """The node name and parsed parameters are handed to the masking helper.""" + extra = { + "instance_info": json.dumps({"kernel_append_params": "pw=hunter2"}), + "netplan_parameters": json.dumps({"eth0": "dhcp"}), + "frr_parameters": json.dumps({"asn": "65000"}), + } + mock_conn.baremetal.get_node.return_value = _node(extra, name="node-7") + + openstack_tasks.get_baremetal_node_parameters("u1") + + stub_mask.assert_called_once_with( + "node-7", "pw=hunter2", {"eth0": "dhcp"}, {"asn": "65000"} + ) + + +def test_node_parameters_returns_masked_values(mock_conn, mocker): + """The masking helper's output is what the task returns.""" + mocker.patch( + "osism.tasks.openstack._mask_node_secret_parameters", + return_value=("masked-kernel", {"net": "masked"}, {"frr": "masked"}), + ) + extra = {"instance_info": json.dumps({"kernel_append_params": "raw"})} + mock_conn.baremetal.get_node.return_value = _node(extra) + + result = openstack_tasks.get_baremetal_node_parameters("u1") + + assert result == { + "kernel_append_params": "masked-kernel", + "netplan_parameters": {"net": "masked"}, + "frr_parameters": {"frr": "masked"}, + } + + +# --------------------------------------------------------------------------- +# _mask_node_secret_parameters (masking helper) +# --------------------------------------------------------------------------- +# +# This is the single seam onto ``osism.tasks.conductor.utils``; the conductor +# internals (``get_vault``, ``deep_decrypt``, ``mask_secrets``) are patched at +# source only here. + + +def test_mask_without_netbox(no_netbox): + """Without a NetBox client no masking happens and inputs pass through.""" + result = openstack_tasks._mask_node_secret_parameters( + "node-1", "pw=hunter2", {}, {} + ) + + assert result == ("pw=hunter2", {}, {}) + + +def test_mask_without_node_name(mock_nb): + """A missing node name skips the NetBox lookup and returns raw values.""" + result = openstack_tasks._mask_node_secret_parameters(None, "pw=hunter2", {}, {}) + + assert result == ("pw=hunter2", {}, {}) + mock_nb.dcim.devices.get.assert_not_called() + + +def test_mask_secrets_none_treated_as_empty(node_params_env): + """A ``secrets`` custom field of ``None`` behaves like an empty dict.""" + node_params_env.device.custom_fields["secrets"] = None + + kernel, _, _ = openstack_tasks._mask_node_secret_parameters( + "node-1", "pw=hunter2", {}, {} + ) + + assert kernel == "pw=hunter2" + node_params_env.deep_decrypt.assert_called_once_with( + {}, node_params_env.get_vault.return_value + ) + + +def test_mask_masks_secret_values(node_params_env): + """Stripped values of password/secret keys are replaced by ``***``; other + values stay visible.""" + node_params_env.device.custom_fields["secrets"] = { + "os_password": " hunter2 ", + "api_secret": "tok123", + "plain": "visible", + } + + kernel, _, _ = openstack_tasks._mask_node_secret_parameters( + "node-1", "pw=hunter2 tok=tok123 keep=visible", {}, {} + ) + + assert kernel == "pw=*** tok=*** keep=visible" + + +def test_mask_ironic_osism_param_by_name(node_params_env): + """The ``ironic_osism_aa`` param name is collected before decryption, so + the ``osism-aa=...`` argument is masked by the name-based regex even when + vault decryption drops the key and its value cannot be collected.""" + node_params_env.device.custom_fields["secrets"] = {"ironic_osism_aa": "topsecret"} + node_params_env.deep_decrypt.side_effect = lambda secrets, vault: secrets.pop( + "ironic_osism_aa", None + ) + + kernel, _, _ = openstack_tasks._mask_node_secret_parameters( + "node-1", "osism-aa=topsecret other=x", {}, {} + ) + + assert kernel == "osism-aa=*** other=x" + + +def test_mask_ignores_non_string_secret_values(node_params_env): + node_params_env.device.custom_fields["secrets"] = {"db_password": 42} + + kernel, _, _ = openstack_tasks._mask_node_secret_parameters( + "node-1", "pw=42", {}, {} + ) + + assert kernel == "pw=42" + + +def test_mask_forwards_secret_values_to_mask_secrets(mocker, node_params_env): + """The collected secret values reach ``mask_secrets`` for both parameter + dicts (frr first, then netplan) with the ``***`` mask.""" + node_params_env.device.custom_fields["secrets"] = {"os_password": "hunter2"} + mask = mocker.patch( + "osism.tasks.conductor.utils.mask_secrets", return_value={"masked": True} + ) + + _, netplan, frr = openstack_tasks._mask_node_secret_parameters( + "node-1", "pw=hunter2", {"eth0": "dhcp"}, {"asn": "65000"} + ) + + assert mask.call_args_list == [ + call({"asn": "65000"}, mask="***", secret_values={"hunter2"}), + call({"eth0": "dhcp"}, mask="***", secret_values={"hunter2"}), + ] + assert frr == {"masked": True} + assert netplan == {"masked": True} + + +def test_mask_device_lookup_fallback(node_params_env): + """No direct name match: the ``cf_inventory_hostname`` filter hit is used.""" + node_params_env.nb.dcim.devices.get.return_value = None + device = SimpleNamespace(custom_fields={"secrets": {"os_password": "hunter2"}}) + node_params_env.nb.dcim.devices.filter.return_value = [device] + + kernel, _, _ = openstack_tasks._mask_node_secret_parameters( + "node-1", "pw=hunter2", {}, {} + ) + + node_params_env.nb.dcim.devices.filter.assert_called_once_with( + cf_inventory_hostname="node-1" + ) + assert kernel == "pw=***" + + +def test_mask_error_returns_unmasked(node_params_env, loguru_logs): + """A failing NetBox/vault path only logs a debug message; the parameters + are returned unmasked.""" + node_params_env.nb.dcim.devices.get.side_effect = Exception("netbox down") + + kernel, _, _ = openstack_tasks._mask_node_secret_parameters( + "node-1", "pw=hunter2", {}, {} + ) + + assert kernel == "pw=hunter2" + assert _has_log(loguru_logs, "DEBUG", "Could not mask secrets") + + +# --------------------------------------------------------------------------- +# Thin Celery task wrappers (parametrized) +# --------------------------------------------------------------------------- + +# Each wrapper fetches a connection and delegates to exactly one SDK method, +# propagating its return value: (task name, call args, call kwargs, SDK method +# path on the connection, expected args, expected kwargs). +THIN_WRAPPER_VARIANTS = [ + pytest.param( + "image_get", + ("cirros",), + {}, + "image.find_image", + ("cirros",), + {}, + id="image_get", + ), + pytest.param( + "network_get", + ("net1",), + {}, + "network.find_network", + ("net1",), + {}, + id="network_get", + ), + pytest.param( + "baremetal_node_create", + ("node1",), + {}, + "baremetal.create_node", + (), + {"name": "node1"}, + id="node_create_default", + ), + pytest.param( + "baremetal_node_create", + ("node1", {"driver": "redfish"}), + {}, + "baremetal.create_node", + (), + {"driver": "redfish", "name": "node1"}, + id="node_create_attributes", + ), + pytest.param( + "baremetal_node_delete", + ("n1",), + {}, + "baremetal.delete_node", + ("n1",), + {}, + id="node_delete", + ), + pytest.param( + "baremetal_node_update", + ("n1",), + {}, + "baremetal.update_node", + ("n1",), + {}, + id="node_update_default", + ), + pytest.param( + "baremetal_node_update", + ("n1", {"maintenance": True}), + {}, + "baremetal.update_node", + ("n1",), + {"maintenance": True}, + id="node_update_attributes", + ), + pytest.param( + "baremetal_node_show", + ("n1",), + {}, + "baremetal.find_node", + ("n1", False), + {}, + id="node_show_default", + ), + pytest.param( + "baremetal_node_show", + ("n1",), + {"ignore_missing": True}, + "baremetal.find_node", + ("n1", True), + {}, + id="node_show_ignore_missing", + ), + pytest.param( + "baremetal_node_validate", + ("n1",), + {}, + "baremetal.validate_node", + ("n1",), + {"required": ()}, + id="node_validate", + ), + pytest.param( + "baremetal_node_set_provision_state", + ("n1", "active"), + {}, + "baremetal.set_node_provision_state", + ("n1", "active"), + {}, + id="node_set_provision_state", + ), + pytest.param( + "baremetal_port_create", + ({"address": "aa:bb"},), + {}, + "baremetal.create_port", + (), + {"address": "aa:bb"}, + id="port_create", + ), + pytest.param( + "baremetal_port_delete", + ("p1",), + {}, + "baremetal.delete_port", + ("p1",), + {}, + id="port_delete", + ), +] + + +@pytest.mark.parametrize( + "task_name, args, kwargs, method_path, expected_args, expected_kwargs", + THIN_WRAPPER_VARIANTS, +) +def test_thin_wrapper_delegates( + mock_conn, task_name, args, kwargs, method_path, expected_args, expected_kwargs +): + task = getattr(openstack_tasks, task_name) + method = attrgetter(method_path)(mock_conn) + + result = task.__wrapped__(*args, **kwargs) + + method.assert_called_once_with(*expected_args, **expected_kwargs) + assert result is method.return_value + + +def test_baremetal_node_list_materializes_generator(mock_conn): + nodes = [SimpleNamespace(name="a"), SimpleNamespace(name="b")] + mock_conn.baremetal.nodes.return_value = iter(nodes) + + result = openstack_tasks.baremetal_node_list.__wrapped__() + + mock_conn.baremetal.nodes.assert_called_once_with() + assert result == nodes + + +def test_baremetal_port_list_materializes_generator_with_defaults(mock_conn): + ports = [SimpleNamespace(id="p1")] + mock_conn.baremetal.ports.return_value = iter(ports) + + result = openstack_tasks.baremetal_port_list.__wrapped__() + + mock_conn.baremetal.ports.assert_called_once_with(details=False) + assert result == ports + + +def test_baremetal_port_list_forwards_attributes(mock_conn): + mock_conn.baremetal.ports.return_value = iter([]) + + openstack_tasks.baremetal_port_list.__wrapped__( + details=True, attributes={"node_uuid": "n1"} + ) + + mock_conn.baremetal.ports.assert_called_once_with(details=True, node_uuid="n1") + + +def test_wait_for_nodes_provision_state_returns_first_result(mock_conn): + mock_conn.baremetal.wait_for_nodes_provision_state.return_value = [ + "first", + "second", + ] + + result = openstack_tasks.baremetal_node_wait_for_nodes_provision_state.__wrapped__( + "n1", "active" + ) + + mock_conn.baremetal.wait_for_nodes_provision_state.assert_called_once_with( + ["n1"], "active" + ) + assert result == "first" + + +def test_wait_for_nodes_provision_state_empty_result(mock_conn): + mock_conn.baremetal.wait_for_nodes_provision_state.return_value = [] + + result = openstack_tasks.baremetal_node_wait_for_nodes_provision_state.__wrapped__( + "n1", "active" + ) + + assert result is None + + +def test_set_boot_device_defaults_to_non_persistent(mock_conn): + result = openstack_tasks.baremetal_node_set_boot_device.__wrapped__("n1", "pxe") + + mock_conn.baremetal.set_node_boot_device.assert_called_once_with( + "n1", "pxe", persistent=False + ) + assert result is None + + +def test_set_boot_device_forwards_persistent(mock_conn): + openstack_tasks.baremetal_node_set_boot_device.__wrapped__( + "n1", "disk", persistent=True + ) + + mock_conn.baremetal.set_node_boot_device.assert_called_once_with( + "n1", "disk", persistent=True + ) + + +def test_set_power_state_without_wait_returns_node(mock_conn): + result = openstack_tasks.baremetal_node_set_power_state.__wrapped__( + "n1", "power on" + ) + + mock_conn.baremetal.set_node_power_state.assert_called_once_with("n1", "power on") + mock_conn.baremetal.get_node.assert_called_once_with("n1") + mock_conn.baremetal.wait_for_node_power_state.assert_not_called() + assert result is mock_conn.baremetal.get_node.return_value + + +def test_set_power_state_with_wait(mock_conn): + result = openstack_tasks.baremetal_node_set_power_state.__wrapped__( + "n1", "power on", wait=True, timeout=30 + ) + + mock_conn.baremetal.wait_for_node_power_state.assert_called_once_with( + "n1", "power on", timeout=30 + ) + mock_conn.baremetal.get_node.assert_not_called() + assert result is mock_conn.baremetal.wait_for_node_power_state.return_value + + +def test_set_target_raid_config(mock_conn): + """The node's RAID state endpoint is PUT with the pinned microversion.""" + mock_conn.baremetal.get_node.return_value = {"uuid": "u1"} + mock_conn.baremetal.put.return_value = SimpleNamespace(ok=True, content=b"payload") + + result = openstack_tasks.baremetal_node_set_target_raid_config.__wrapped__( + "n1", {"logical_disks": []} + ) + + mock_conn.baremetal.get_node.assert_called_once_with("n1") + mock_conn.baremetal.put.assert_called_once_with( + "/nodes/u1/states/raid", microversion="1.12", json={"logical_disks": []} + ) + assert result == (True, b"payload") + + +def test_setup_periodic_tasks_is_noop(mocker): + sender = mocker.MagicMock() + + openstack_tasks.setup_periodic_tasks(sender) + + assert sender.mock_calls == [] diff --git a/tests/unit/tasks/test_openstack_env.py b/tests/unit/tasks/test_openstack_env.py new file mode 100644 index 00000000..2afb15ca --- /dev/null +++ b/tests/unit/tasks/test_openstack_env.py @@ -0,0 +1,588 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the cloud credential/environment helpers of +``osism/tasks/openstack.py``. + +Covers ``get_cloud_password``, ``get_cloud_helpers``, +``setup_cloud_environment``, ``cleanup_cloud_environment``, the module-level +two-argument ``get_openstack_connection`` and ``run_openstack_command_with_cloud``. + +Every external effect is mocked: the environment helpers never touch the real +filesystem or working directory because the module-level ``os``/``shutil``/ +``yaml`` name bindings (plus ``builtins.open``) are patched (see the shared +``mock_os`` fixture in ``conftest.py``). The module-level +``get_openstack_connection(cloud, password)`` is exercised against a patched +``openstack.connect``; ``openstacksdk`` and ``keystoneauth1`` are pinned runtime +dependencies, so their real exception classes are raised instead of stand-ins. +""" + +from types import SimpleNamespace +from unittest.mock import call + +import keystoneauth1.exceptions +import openstack.exceptions +import pytest + +from osism.tasks import openstack as openstack_tasks + +SECRETS_PATH = "/opt/configuration/environments/openstack/secrets.yml" + +Unauthorized = keystoneauth1.exceptions.http.Unauthorized +SDKException = openstack.exceptions.SDKException + + +def _missing_options(*dests): + """Build a real ``MissingRequiredOptions`` whose message names ``dests``.""" + return keystoneauth1.exceptions.auth_plugins.MissingRequiredOptions( + [SimpleNamespace(dest=dest) for dest in dests] + ) + + +def _has_log(records, level, substring): + return any(r["level"] == level and substring in r["message"] for r in records) + + +def _exists_only(fake_os, *paths): + """Make the patched ``os.path.exists`` report exactly ``paths`` as present.""" + existing = set(paths) + fake_os.path.exists.side_effect = lambda path: path in existing + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def secrets(mocker, mock_os): + """An existing secrets file whose parsed content the test controls.""" + mock_os.path.exists.return_value = True + return mocker.patch("osism.tasks.openstack.load_yaml_file") + + +@pytest.fixture +def cloud_env(mocker, mock_os): + """Mock every external effect of ``setup_cloud_environment``. + + By default no path exists, ``get_cloud_password`` yields ``"hunter2"`` + (password branch) and ``open``/``yaml``/``shutil`` are inert mocks. + """ + get_cloud_password = mocker.patch( + "osism.tasks.openstack.get_cloud_password", return_value="hunter2" + ) + fake_shutil = mocker.patch("osism.tasks.openstack.shutil") + fake_yaml = mocker.patch("osism.tasks.openstack.yaml") + open_ = mocker.patch("builtins.open", mocker.mock_open()) + return SimpleNamespace( + os=mock_os, + get_cloud_password=get_cloud_password, + shutil=fake_shutil, + yaml=fake_yaml, + open=open_, + ) + + +@pytest.fixture +def mock_connect(mocker, mock_os): + """Patch ``openstack.connect`` at source; no secure.yml exists by default.""" + return mocker.patch("openstack.connect") + + +@pytest.fixture +def command_env(mocker): + """Mock the collaborators of ``run_openstack_command_with_cloud``.""" + setup = mocker.patch( + "osism.tasks.openstack.setup_cloud_environment", + return_value=("pw", ["/tmp/clouds.yaml"], "/orig", True), + ) + cleanup = mocker.patch("osism.tasks.openstack.cleanup_cloud_environment") + run = mocker.patch("osism.tasks.openstack.run_command", return_value=0) + return SimpleNamespace(setup=setup, cleanup=cleanup, run=run) + + +# --------------------------------------------------------------------------- +# get_cloud_password +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("cloud", [None, ""], ids=["none", "empty"]) +def test_get_cloud_password_requires_cloud(mocker, loguru_logs, cloud): + """A falsy cloud short-circuits with a warning before any file access.""" + load = mocker.patch("osism.tasks.openstack.load_yaml_file") + + assert openstack_tasks.get_cloud_password(cloud) is None + + load.assert_not_called() + assert _has_log(loguru_logs, "WARNING", "No cloud parameter provided") + + +def test_get_cloud_password_missing_secrets_file(mocker, mock_os, loguru_logs): + """A missing secrets file yields ``None`` without parsing anything.""" + load = mocker.patch("osism.tasks.openstack.load_yaml_file") + + assert openstack_tasks.get_cloud_password("admin") is None + + mock_os.path.exists.assert_called_once_with(SECRETS_PATH) + load.assert_not_called() + assert _has_log(loguru_logs, "WARNING", "Secrets file not found") + + +@pytest.mark.parametrize("content", [None, ["not", "a", "dict"]], ids=["none", "list"]) +def test_get_cloud_password_invalid_secrets_content(secrets, loguru_logs, content): + """``None`` or non-dict content from the YAML loader is rejected.""" + secrets.return_value = content + + assert openstack_tasks.get_cloud_password("admin") is None + + assert _has_log(loguru_logs, "WARNING", "Empty or invalid secrets file") + + +def test_get_cloud_password_normalizes_hyphens(secrets): + """Hyphens in the cloud name map to underscores in the password key.""" + secrets.return_value = {"os_password_admin_system": "pw"} + + assert openstack_tasks.get_cloud_password("admin-system") == "pw" + + secrets.assert_called_once_with(SECRETS_PATH) + + +def test_get_cloud_password_rejects_non_identifier_key(secrets, loguru_logs): + """A cloud name producing a non-identifier key is refused with an error.""" + secrets.return_value = {"os_password_admin": "pw"} + + assert openstack_tasks.get_cloud_password("admin system") is None + + assert _has_log(loguru_logs, "ERROR", "Invalid password key format") + + +def test_get_cloud_password_key_not_found(secrets, loguru_logs): + """A missing password key is only a debug event, not a warning.""" + secrets.return_value = {"os_password_other": "pw"} + + assert openstack_tasks.get_cloud_password("admin") is None + + assert _has_log(loguru_logs, "DEBUG", "not found") + assert not any(r["level"] in ("WARNING", "ERROR") for r in loguru_logs) + + +def test_get_cloud_password_strips_whitespace(secrets): + secrets.return_value = {"os_password_admin": " hunter2 "} + + assert openstack_tasks.get_cloud_password("admin") == "hunter2" + + +def test_get_cloud_password_coerces_to_string(secrets): + """Non-string values (e.g. YAML integers) are coerced via ``str``.""" + secrets.return_value = {"os_password_admin": 42} + + assert openstack_tasks.get_cloud_password("admin") == "42" + + +@pytest.mark.parametrize("value", ["", " "], ids=["empty", "whitespace"]) +def test_get_cloud_password_empty_value(secrets, loguru_logs, value): + """Values that are empty after stripping yield ``None`` with a warning.""" + secrets.return_value = {"os_password_admin": value} + + assert openstack_tasks.get_cloud_password("admin") is None + + assert _has_log(loguru_logs, "WARNING", "empty after conversion") + + +def test_get_cloud_password_load_error(secrets, loguru_logs): + """Loader exceptions are swallowed and logged as errors.""" + secrets.side_effect = Exception("vault broken") + + assert openstack_tasks.get_cloud_password("admin") is None + + assert _has_log(loguru_logs, "ERROR", "Failed to load/decrypt password") + + +# --------------------------------------------------------------------------- +# get_cloud_helpers +# --------------------------------------------------------------------------- + + +def test_get_cloud_helpers_returns_helper_triple(): + assert openstack_tasks.get_cloud_helpers() == ( + openstack_tasks.setup_cloud_environment, + openstack_tasks.get_openstack_connection, + openstack_tasks.cleanup_cloud_environment, + ) + + +# --------------------------------------------------------------------------- +# setup_cloud_environment +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("cloud", [None, ""], ids=["none", "empty"]) +def test_setup_cloud_environment_requires_cloud(cloud_env, loguru_logs, cloud): + """A falsy cloud fails the setup before any password lookup.""" + result = openstack_tasks.setup_cloud_environment(cloud) + + assert result == (None, [], "/orig", False) + cloud_env.get_cloud_password.assert_not_called() + assert _has_log(loguru_logs, "WARNING", "No cloud parameter provided") + + +def test_setup_cloud_environment_injects_password(cloud_env): + """Password branch: the password is injected into the cloud profile, the + combined config plus an empty secure.yml land in /tmp and the working + directory changes there.""" + _exists_only(cloud_env.os, "/etc/openstack/clouds.yaml") + clouds_config = {"clouds": {"admin": {"auth": {"username": "u"}}}} + cloud_env.yaml.safe_load.return_value = clouds_config + + result = openstack_tasks.setup_cloud_environment("admin") + + handle = cloud_env.open.return_value + cloud_env.open.assert_any_call("/etc/openstack/clouds.yaml", "r") + cloud_env.open.assert_any_call("/tmp/clouds.yaml", "w") + cloud_env.open.assert_any_call("/tmp/secure.yml", "w") + cloud_env.yaml.safe_load.assert_called_once_with(handle) + assert clouds_config["clouds"]["admin"]["auth"]["password"] == "hunter2" + assert cloud_env.yaml.dump.call_args_list == [ + call(clouds_config, handle), + call({}, handle), + ] + cloud_env.os.chdir.assert_called_once_with("/tmp") + assert result == ("hunter2", ["/tmp/clouds.yaml", "/tmp/secure.yml"], "/orig", True) + + +def test_setup_cloud_environment_clouds_yml_variant(cloud_env): + """Only ``clouds.yml`` present: the /tmp copy keeps the ``.yml`` suffix.""" + _exists_only(cloud_env.os, "/etc/openstack/clouds.yml") + cloud_env.yaml.safe_load.return_value = {"clouds": {"admin": {"auth": {}}}} + + result = openstack_tasks.setup_cloud_environment("admin") + + cloud_env.open.assert_any_call("/etc/openstack/clouds.yml", "r") + cloud_env.open.assert_any_call("/tmp/clouds.yml", "w") + assert result == ("hunter2", ["/tmp/clouds.yml", "/tmp/secure.yml"], "/orig", True) + + +def test_setup_cloud_environment_creates_auth_section(cloud_env): + """A cloud profile without an ``auth`` key gets one before injection.""" + _exists_only(cloud_env.os, "/etc/openstack/clouds.yaml") + clouds_config = {"clouds": {"admin": {}}} + cloud_env.yaml.safe_load.return_value = clouds_config + + result = openstack_tasks.setup_cloud_environment("admin") + + assert clouds_config["clouds"]["admin"]["auth"] == {"password": "hunter2"} + assert result[3] is True + + +def test_setup_cloud_environment_cloud_missing_from_config(cloud_env, loguru_logs): + """A cloud profile absent from the clouds config fails the setup.""" + _exists_only(cloud_env.os, "/etc/openstack/clouds.yaml") + cloud_env.yaml.safe_load.return_value = {"clouds": {"other": {}}} + + result = openstack_tasks.setup_cloud_environment("admin") + + assert result == (None, [], "/orig", False) + cloud_env.os.chdir.assert_not_called() + assert _has_log(loguru_logs, "WARNING", "not found in clouds config") + + +def test_setup_cloud_environment_without_clouds_file(cloud_env): + """No clouds.yaml at all: the password is still returned for direct SDK + usage, without any temp files or directory change.""" + result = openstack_tasks.setup_cloud_environment("admin") + + assert result == ("hunter2", [], "/orig", True) + cloud_env.os.chdir.assert_not_called() + cloud_env.open.assert_not_called() + + +def test_setup_cloud_environment_write_error(cloud_env, loguru_logs): + """Errors while reading/writing the config fail the setup.""" + _exists_only(cloud_env.os, "/etc/openstack/clouds.yaml") + cloud_env.yaml.safe_load.side_effect = Exception("bad yaml") + + result = openstack_tasks.setup_cloud_environment("admin") + + assert result == (None, [], "/orig", False) + assert _has_log(loguru_logs, "ERROR", "Failed to set up cloud environment") + + +def test_setup_cloud_environment_secure_yml_fallback(cloud_env): + """Fallback branch: without a password, clouds.yaml and secure.yml are + copied to /tmp and the working directory changes there.""" + cloud_env.get_cloud_password.return_value = None + _exists_only( + cloud_env.os, "/etc/openstack/secure.yml", "/etc/openstack/clouds.yaml" + ) + + result = openstack_tasks.setup_cloud_environment("admin") + + assert cloud_env.shutil.copy2.call_args_list == [ + call("/etc/openstack/clouds.yaml", "/tmp/clouds.yaml"), + call("/etc/openstack/secure.yml", "/tmp/secure.yml"), + ] + cloud_env.os.chdir.assert_called_once_with("/tmp") + assert result == (None, ["/tmp/clouds.yaml", "/tmp/secure.yml"], "/orig", True) + + +def test_setup_cloud_environment_secure_yaml_variant(cloud_env): + """Only ``secure.yaml`` present: the /tmp copy keeps the ``.yaml`` suffix.""" + cloud_env.get_cloud_password.return_value = None + _exists_only(cloud_env.os, "/etc/openstack/secure.yaml") + + result = openstack_tasks.setup_cloud_environment("admin") + + cloud_env.shutil.copy2.assert_called_once_with( + "/etc/openstack/secure.yaml", "/tmp/secure.yaml" + ) + assert result == (None, ["/tmp/secure.yaml"], "/orig", True) + + +def test_setup_cloud_environment_no_credentials(cloud_env, loguru_logs): + """Neither a password nor a secure file: the error names the expected + ``os_password_`` key with hyphens normalized.""" + cloud_env.get_cloud_password.return_value = None + + result = openstack_tasks.setup_cloud_environment("admin-system") + + assert result == (None, [], "/orig", False) + assert _has_log(loguru_logs, "ERROR", "os_password_admin_system") + + +def test_setup_cloud_environment_copy_error(cloud_env, loguru_logs): + """A failing copy in the fallback branch fails the setup.""" + cloud_env.get_cloud_password.return_value = None + _exists_only(cloud_env.os, "/etc/openstack/secure.yml") + cloud_env.shutil.copy2.side_effect = OSError("disk full") + + result = openstack_tasks.setup_cloud_environment("admin") + + assert result == (None, [], "/orig", False) + assert _has_log(loguru_logs, "ERROR", "Failed to copy config files") + + +# --------------------------------------------------------------------------- +# cleanup_cloud_environment +# --------------------------------------------------------------------------- + + +def test_cleanup_restores_cwd_and_removes_files(mock_os): + mock_os.path.exists.return_value = True + + openstack_tasks.cleanup_cloud_environment(["/tmp/a", "/tmp/b"], "/orig") + + mock_os.chdir.assert_called_once_with("/orig") + assert mock_os.remove.call_args_list == [call("/tmp/a"), call("/tmp/b")] + + +def test_cleanup_survives_chdir_failure(mock_os, loguru_logs): + """A failing ``chdir`` is only a warning; files are still removed.""" + mock_os.path.exists.return_value = True + mock_os.chdir.side_effect = OSError("gone") + + openstack_tasks.cleanup_cloud_environment(["/tmp/a"], "/orig") + + mock_os.remove.assert_called_once_with("/tmp/a") + assert _has_log( + loguru_logs, "WARNING", "Could not restore original working directory" + ) + + +def test_cleanup_skips_missing_files(mock_os): + mock_os.path.exists.return_value = False + + openstack_tasks.cleanup_cloud_environment(["/tmp/a"], "/orig") + + mock_os.remove.assert_not_called() + + +def test_cleanup_continues_after_remove_failure(mock_os, loguru_logs): + """A failing removal is only a warning; remaining files are processed.""" + mock_os.path.exists.return_value = True + mock_os.remove.side_effect = [OSError("busy"), None] + + openstack_tasks.cleanup_cloud_environment(["/tmp/a", "/tmp/b"], "/orig") + + assert mock_os.remove.call_args_list == [call("/tmp/a"), call("/tmp/b")] + assert _has_log(loguru_logs, "WARNING", "Could not remove temporary file /tmp/a") + + +# --------------------------------------------------------------------------- +# get_openstack_connection (module-level two-argument helper) +# --------------------------------------------------------------------------- + + +def test_connection_with_password(mocker, mock_connect): + """With a password, ``connect`` gets the auth override and the connection + is verified by touching ``current_project``.""" + conn = mocker.MagicMock() + project = mocker.PropertyMock(return_value="proj") + type(conn).current_project = project + mock_connect.return_value = conn + + result = openstack_tasks.get_openstack_connection("admin", password="pw") + + mock_connect.assert_called_once_with(cloud="admin", auth={"password": "pw"}) + project.assert_called_once_with() + assert result is conn + + +def test_connection_password_fallback_to_secure_yml(mocker, mock_connect, mock_os): + """``Unauthorized`` with an existing secure.yml retries without the + password override.""" + mock_os.path.exists.side_effect = lambda p: p == "/etc/openstack/secure.yml" + fallback_conn = mocker.MagicMock() + mock_connect.side_effect = [Unauthorized(), fallback_conn] + + result = openstack_tasks.get_openstack_connection("admin", password="wrong") + + assert result is fallback_conn + assert mock_connect.call_args_list == [ + call(cloud="admin", auth={"password": "wrong"}), + call(cloud="admin"), + ] + + +def test_connection_password_unauthorized_without_secure_yml(mock_connect, loguru_logs): + """``Unauthorized`` without a secure.yml exits; the error names the + normalized password key.""" + mock_connect.side_effect = Unauthorized() + + with pytest.raises(SystemExit) as excinfo: + openstack_tasks.get_openstack_connection("admin-system", password="wrong") + + assert excinfo.value.code == 1 + assert _has_log(loguru_logs, "ERROR", "os_password_admin_system") + + +def test_connection_password_missing_options(mock_connect, loguru_logs): + mock_connect.side_effect = _missing_options("auth_url") + + with pytest.raises(SystemExit): + openstack_tasks.get_openstack_connection("admin", password="pw") + + assert _has_log(loguru_logs, "ERROR", "Missing configuration for cloud 'admin'") + + +def test_connection_password_sdk_error(mock_connect, loguru_logs): + mock_connect.side_effect = SDKException("boom") + + with pytest.raises(SystemExit): + openstack_tasks.get_openstack_connection("admin", password="pw") + + assert _has_log(loguru_logs, "ERROR", "OpenStack SDK error") + + +def test_connection_without_password(mocker, mock_connect): + conn = mocker.MagicMock() + mock_connect.return_value = conn + + result = openstack_tasks.get_openstack_connection("admin") + + mock_connect.assert_called_once_with(cloud="admin") + assert result is conn + + +def test_connection_without_password_missing_password_option(mock_connect, loguru_logs): + """A ``MissingRequiredOptions`` naming the password produces the specific + "No password configured" error.""" + mock_connect.side_effect = _missing_options("password") + + with pytest.raises(SystemExit): + openstack_tasks.get_openstack_connection("admin-system") + + assert _has_log( + loguru_logs, "ERROR", "No password configured for cloud 'admin-system'" + ) + assert _has_log(loguru_logs, "ERROR", "os_password_admin_system") + + +def test_connection_without_password_missing_other_option(mock_connect, loguru_logs): + """Other missing options produce the generic configuration error.""" + mock_connect.side_effect = _missing_options("auth_url") + + with pytest.raises(SystemExit): + openstack_tasks.get_openstack_connection("admin") + + assert _has_log(loguru_logs, "ERROR", "Missing configuration for cloud 'admin'") + + +def test_connection_without_password_unauthorized(mock_connect, loguru_logs): + mock_connect.side_effect = Unauthorized() + + with pytest.raises(SystemExit): + openstack_tasks.get_openstack_connection("admin") + + assert _has_log(loguru_logs, "ERROR", "Authentication failed for cloud 'admin'") + + +def test_connection_without_password_sdk_error(mock_connect, loguru_logs): + mock_connect.side_effect = SDKException("boom") + + with pytest.raises(SystemExit): + openstack_tasks.get_openstack_connection("admin") + + assert _has_log(loguru_logs, "ERROR", "OpenStack SDK error") + + +def test_connection_test_failure_after_connect(mocker, mock_connect): + """``connect`` succeeds but the ``current_project`` probe fails.""" + conn = mocker.MagicMock() + type(conn).current_project = mocker.PropertyMock( + side_effect=SDKException("no project") + ) + mock_connect.return_value = conn + + with pytest.raises(SystemExit): + openstack_tasks.get_openstack_connection("admin", password="pw") + + +# --------------------------------------------------------------------------- +# run_openstack_command_with_cloud +# --------------------------------------------------------------------------- + + +def test_run_openstack_command_with_cloud_delegates(command_env): + result = openstack_tasks.run_openstack_command_with_cloud( + "req-1", + "/usr/local/bin/x", + "admin", + ["--a", "--b"], + publish=False, + locking=True, + auto_release_time=60, + ) + + command_env.setup.assert_called_once_with("admin") + command_env.run.assert_called_once_with( + "req-1", + "/usr/local/bin/x", + {}, + "--a", + "--b", + publish=False, + locking=True, + auto_release_time=60, + ignore_env=True, + ) + command_env.cleanup.assert_called_once_with(["/tmp/clouds.yaml"], "/orig") + assert result == 0 + + +def test_run_openstack_command_with_cloud_cleans_up_on_error(command_env): + """The temp files from setup are cleaned up even when the command fails.""" + command_env.run.side_effect = RuntimeError("worker died") + + with pytest.raises(RuntimeError): + openstack_tasks.run_openstack_command_with_cloud("req-1", "/bin/x", "admin", []) + + command_env.cleanup.assert_called_once_with(["/tmp/clouds.yaml"], "/orig") + + +def test_run_openstack_command_with_cloud_runs_despite_setup_failure(command_env): + """Current behavior: a failed setup does not short-circuit the command.""" + command_env.setup.return_value = (None, [], "/orig", False) + + openstack_tasks.run_openstack_command_with_cloud("req-1", "/bin/x", "admin", []) + + command_env.run.assert_called_once() + command_env.cleanup.assert_called_once_with([], "/orig") diff --git a/tests/unit/tasks/test_openstack_managers.py b/tests/unit/tasks/test_openstack_managers.py new file mode 100644 index 00000000..042eebf4 --- /dev/null +++ b/tests/unit/tasks/test_openstack_managers.py @@ -0,0 +1,258 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the manager tasks of ``osism/tasks/openstack.py``. + +Covers ``image_manager``, ``flavor_manager``, ``project_manager`` and +``project_manager_sync``. + +Every external effect is mocked: the cloud environment helpers, ``run_command`` +and the task-lock check are patched, and the module-level ``os`` binding comes +from the shared ``mock_os`` fixture (see ``conftest.py``) so no test touches the +real filesystem or working directory. The bound tasks (``bind=True``) are +exercised through ``task.__wrapped__(...)`` (already bound to the task instance, +so ``self.request.id`` is ``None``). +""" + +import pathlib + +import pytest + +from osism.tasks import openstack as openstack_tasks + +# --------------------------------------------------------------------------- +# image_manager +# --------------------------------------------------------------------------- + + +def test_image_manager_delegates_without_configs(mocker): + """Without configs the task delegates to the generalized helper.""" + check = mocker.patch("osism.tasks.openstack.utils.check_task_lock_and_exit") + delegate = mocker.patch( + "osism.tasks.openstack.run_openstack_command_with_cloud", return_value="RC" + ) + + result = openstack_tasks.image_manager.__wrapped__( + "--dry-run", cloud="admin", publish=False, locking=True, auto_release_time=60 + ) + + check.assert_called_once_with() + delegate.assert_called_once_with( + None, + "/usr/local/bin/openstack-image-manager", + "admin", + ("--dry-run",), + publish=False, + locking=True, + auto_release_time=60, + ) + assert result == "RC" + + +def test_image_manager_rewrites_images_arguments(mocker): + """With configs, each config lands as a file in a temp directory, both + ``--images=...`` and ``--images `` arguments are dropped and the + temp directory is appended as the new ``--images`` value.""" + mocker.patch("osism.tasks.openstack.utils.check_task_lock_and_exit") + setup = mocker.patch( + "osism.tasks.openstack.setup_cloud_environment", + return_value=(None, [], "/orig", True), + ) + cleanup = mocker.patch("osism.tasks.openstack.cleanup_cloud_environment") + captured = {} + + def fake_run_command(request_id, command, env, *args, **kwargs): + # The temp directory only exists while run_command executes, so its + # contents must be captured here. + images_dir = pathlib.Path(args[args.index("--images") + 1]) + captured["request_id"] = request_id + captured["command"] = command + captured["env"] = env + captured["args"] = args + captured["kwargs"] = kwargs + captured["configs"] = sorted(p.read_text() for p in images_dir.iterdir()) + return "RC" + + mocker.patch("osism.tasks.openstack.run_command", side_effect=fake_run_command) + + result = openstack_tasks.image_manager.__wrapped__( + "--dry-run", + "--images=inline-dir", + "--images", + "pair-dir", + configs=["yaml-a", "yaml-b"], + cloud="admin", + ) + + setup.assert_called_once_with("admin") + assert captured["request_id"] is None + assert captured["command"] == "/usr/local/bin/openstack-image-manager" + assert captured["env"] == {} + assert captured["args"][:2] == ("--dry-run", "--images") + assert "--images=inline-dir" not in captured["args"] + assert "pair-dir" not in captured["args"] + assert captured["configs"] == ["yaml-a", "yaml-b"] + assert captured["kwargs"] == { + "publish": True, + "locking": False, + "auto_release_time": 3600, + "ignore_env": True, + } + cleanup.assert_called_once_with([], "/orig") + assert result == "RC" + + +def test_image_manager_cleans_up_on_run_command_error(mocker): + mocker.patch("osism.tasks.openstack.utils.check_task_lock_and_exit") + mocker.patch( + "osism.tasks.openstack.setup_cloud_environment", + return_value=(None, [], "/orig", True), + ) + cleanup = mocker.patch("osism.tasks.openstack.cleanup_cloud_environment") + mocker.patch( + "osism.tasks.openstack.run_command", side_effect=RuntimeError("worker died") + ) + + with pytest.raises(RuntimeError): + openstack_tasks.image_manager.__wrapped__(configs=["cfg"], cloud="admin") + + cleanup.assert_called_once_with([], "/orig") + + +def test_image_manager_trailing_images_flag_raises(mocker): + """Current behavior: a trailing ``--images`` without a value raises + ``IndexError`` (only ``ValueError`` is caught); cleanup still runs.""" + mocker.patch("osism.tasks.openstack.utils.check_task_lock_and_exit") + mocker.patch( + "osism.tasks.openstack.setup_cloud_environment", + return_value=(None, [], "/orig", True), + ) + cleanup = mocker.patch("osism.tasks.openstack.cleanup_cloud_environment") + run = mocker.patch("osism.tasks.openstack.run_command") + + with pytest.raises(IndexError): + openstack_tasks.image_manager.__wrapped__( + "--images", configs=["cfg"], cloud="admin" + ) + + run.assert_not_called() + cleanup.assert_called_once_with([], "/orig") + + +# --------------------------------------------------------------------------- +# flavor_manager / project_manager / project_manager_sync +# --------------------------------------------------------------------------- + + +def test_flavor_manager_delegates(mocker): + check = mocker.patch("osism.tasks.openstack.utils.check_task_lock_and_exit") + delegate = mocker.patch( + "osism.tasks.openstack.run_openstack_command_with_cloud", return_value="RC" + ) + + result = openstack_tasks.flavor_manager.__wrapped__( + "--x", cloud="admin", publish=False, locking=True, auto_release_time=60 + ) + + check.assert_called_once_with() + delegate.assert_called_once_with( + None, + "/usr/local/bin/openstack-flavor-manager", + "admin", + ("--x",), + publish=False, + locking=True, + auto_release_time=60, + ) + assert result == "RC" + + +PROJECT_MANAGER_VARIANTS = [ + pytest.param( + "project_manager", + "/openstack-project-manager/openstack_project_manager/create.py", + id="create", + ), + pytest.param( + "project_manager_sync", + "/openstack-project-manager/openstack_project_manager/manage.py", + id="manage", + ), +] + + +@pytest.mark.parametrize("task_name, script_path", PROJECT_MANAGER_VARIANTS) +def test_project_manager_delegates(mocker, mock_os, task_name, script_path): + """The script path is prepended to the arguments and the command runs from + the openstack-project-manager checkout.""" + check = mocker.patch("osism.tasks.openstack.utils.check_task_lock_and_exit") + setup = mocker.patch( + "osism.tasks.openstack.setup_cloud_environment", + return_value=("pw", ["/tmp/clouds.yaml"], "/orig", True), + ) + cleanup = mocker.patch("osism.tasks.openstack.cleanup_cloud_environment") + run = mocker.patch("osism.tasks.openstack.run_command", return_value="RC") + + result = getattr(openstack_tasks, task_name).__wrapped__( + "--x", cloud="admin", publish=False + ) + + check.assert_called_once_with() + setup.assert_called_once_with("admin") + mock_os.chdir.assert_called_once_with("/openstack-project-manager") + run.assert_called_once_with( + None, + "/usr/local/bin/python3", + {}, + script_path, + "--x", + publish=False, + locking=False, + auto_release_time=3600, + ignore_env=True, + ) + cleanup.assert_called_once_with(["/tmp/clouds.yaml"], "/orig") + assert result == "RC" + + +@pytest.mark.parametrize("task_name, script_path", PROJECT_MANAGER_VARIANTS) +def test_project_manager_cleans_up_on_error(mocker, mock_os, task_name, script_path): + mocker.patch("osism.tasks.openstack.utils.check_task_lock_and_exit") + mocker.patch( + "osism.tasks.openstack.setup_cloud_environment", + return_value=(None, ["/tmp/clouds.yaml"], "/orig", True), + ) + cleanup = mocker.patch("osism.tasks.openstack.cleanup_cloud_environment") + run = mocker.patch("osism.tasks.openstack.run_command") + mock_os.chdir.side_effect = OSError("missing directory") + + with pytest.raises(OSError): + getattr(openstack_tasks, task_name).__wrapped__(cloud="admin") + + run.assert_not_called() + cleanup.assert_called_once_with(["/tmp/clouds.yaml"], "/orig") + + +MANAGER_LOCK_VARIANTS = [ + pytest.param("image_manager", {}, id="image_manager"), + pytest.param("image_manager", {"configs": ["cfg"]}, id="image_manager_configs"), + pytest.param("flavor_manager", {}, id="flavor_manager"), + pytest.param("project_manager", {}, id="project_manager"), + pytest.param("project_manager_sync", {}, id="project_manager_sync"), +] + + +@pytest.mark.parametrize("task_name, kwargs", MANAGER_LOCK_VARIANTS) +def test_manager_tasks_abort_when_task_lock_active(mocker, task_name, kwargs): + """The task-lock check runs before any cloud setup or delegation.""" + mocker.patch( + "osism.tasks.openstack.utils.check_task_lock_and_exit", + side_effect=SystemExit(1), + ) + delegate = mocker.patch("osism.tasks.openstack.run_openstack_command_with_cloud") + setup = mocker.patch("osism.tasks.openstack.setup_cloud_environment") + + with pytest.raises(SystemExit): + getattr(openstack_tasks, task_name).__wrapped__(cloud="admin", **kwargs) + + delegate.assert_not_called() + setup.assert_not_called()