diff --git a/osism/commands/baremetal.py b/osism/commands/baremetal.py index 972f6b7af..4548edf13 100644 --- a/osism/commands/baremetal.py +++ b/osism/commands/baremetal.py @@ -197,6 +197,7 @@ def take_action(self, parsed_args): return 1 deploy_nodes = [node] + failed = False for node in deploy_nodes: if not node: continue @@ -238,6 +239,7 @@ def take_action(self, parsed_args): logger.warning( f"Node {node.name} ({node.id}) could not be validated" ) + failed = True continue # NOTE: Prepare osism config drive try: @@ -352,6 +354,7 @@ def take_action(self, parsed_args): logger.warning( f"Failed to build config drive for {node.name} ({node.id}): {exc}" ) + failed = True continue node_vendor = node.properties.get("vendor", "").strip().lower() if node_vendor == "supermicro": @@ -399,7 +402,11 @@ def take_action(self, parsed_args): logger.warning( f"Node {node.name} ({node.id}) could not be moved to active state: {exc}" ) + failed = True continue + + if failed: + return 1 finally: cleanup_cloud_environment(temp_files, original_cwd) @@ -563,6 +570,7 @@ def take_action(self, parsed_args): ) except Exception as exc: logger.error(f"Failed to generate playbook for {name}: {exc}") + return 1 finally: cleanup_cloud_environment(temp_files, original_cwd) else: @@ -676,6 +684,7 @@ def take_action(self, parsed_args): ) except Exception as exc: logger.error(f"Failed to generate playbook for {name}: {exc}") + return 1 class BaremetalUndeploy(Command): @@ -748,6 +757,7 @@ def take_action(self, parsed_args): return 1 deploy_nodes = [node] + failed = False for node in deploy_nodes: if not node: continue @@ -784,11 +794,15 @@ def take_action(self, parsed_args): logger.warning( f"Node {node.name} ({node.id}) could not be moved to available state: {exc}" ) + failed = True continue else: logger.warning( f"Node {node.name} ({node.id}) not in supported provision state" ) + + if failed: + return 1 finally: cleanup_cloud_environment(temp_files, original_cwd) @@ -1059,6 +1073,7 @@ def take_action(self, parsed_args): return 1 burn_in_nodes = [node] + failed = False for node in burn_in_nodes: if not node: continue @@ -1076,6 +1091,7 @@ def take_action(self, parsed_args): logger.warning( f"Node {node.name} ({node.id}) could not be moved to manageable state: {exc}" ) + failed = True continue if node.provision_state in ["manageable"]: @@ -1110,6 +1126,7 @@ def take_action(self, parsed_args): logger.warning( f"Burn-In of node {node.name} ({node.id}) failed: {exc}" ) + failed = True continue elif node.provision_state in ["active"]: # NOTE: Use service step to run burn-in @@ -1135,6 +1152,7 @@ def take_action(self, parsed_args): logger.warning( f"Burn-In of node {node.name} ({node.id}) failed: {exc}" ) + failed = True continue else: @@ -1142,6 +1160,9 @@ def take_action(self, parsed_args): f"Node {node.name} ({node.id}) not in supported state! Provision state: {node.provision_state}, maintenance mode: {node['maintenance']}" ) continue + + if failed: + return 1 finally: cleanup_cloud_environment(temp_files, original_cwd) @@ -1216,6 +1237,7 @@ def take_action(self, parsed_args): return 1 clean_nodes = [node] + failed = False for node in clean_nodes: if not node: continue @@ -1239,6 +1261,7 @@ def take_action(self, parsed_args): logger.warning( f"Node {node.name} ({node.id}) could not be moved to manageable state: {exc}" ) + failed = True continue if node.provision_state in ["manageable"]: @@ -1276,11 +1299,15 @@ def take_action(self, parsed_args): logger.warning( f"Clean of node {node.name} ({node.id}) failed: {exc}" ) + failed = True continue else: logger.warning( f"Node {node.name} ({node.id}) not in supported state! Provision state: {node.provision_state}, maintenance mode: {node['maintenance']}" ) + + if failed: + return 1 finally: cleanup_cloud_environment(temp_files, original_cwd) @@ -1340,6 +1367,7 @@ def take_action(self, parsed_args): return 1 provide_nodes = [node] + failed = False for node in provide_nodes: if not node: continue @@ -1354,11 +1382,15 @@ def take_action(self, parsed_args): logger.warning( f"Node {node.name} ({node.id}) could not be moved to available state: {exc}" ) + failed = True continue else: logger.warning( f"Node {node.name} ({node.id}) not in supported state! Provision state: {node.provision_state}, maintenance mode: {node['maintenance']}" ) + + if failed: + return 1 finally: cleanup_cloud_environment(temp_files, original_cwd) @@ -1414,6 +1446,7 @@ def take_action(self, parsed_args): logger.error( f"Setting maintenance mode on node {node.name} ({node.id}) failed: {exc}" ) + return 1 finally: cleanup_cloud_environment(temp_files, original_cwd) @@ -1462,6 +1495,7 @@ def take_action(self, parsed_args): logger.error( f"Unsetting maintenance mode on node {node.name} ({node.id}) failed: {exc}" ) + return 1 finally: cleanup_cloud_environment(temp_files, original_cwd) @@ -1514,6 +1548,7 @@ def take_action(self, parsed_args): logger.info(f"Successfully powered on node {node.name} ({node.id})") except Exception as exc: logger.error(f"Failed to power on node {node.name} ({node.id}): {exc}") + return 1 finally: cleanup_cloud_environment(temp_files, original_cwd) @@ -1576,6 +1611,7 @@ def take_action(self, parsed_args): logger.info(f"Successfully {action} node {node.name} ({node.id})") except Exception as exc: logger.error(f"Failed to power off node {node.name} ({node.id}): {exc}") + return 1 finally: cleanup_cloud_environment(temp_files, original_cwd) @@ -1650,6 +1686,7 @@ def take_action(self, parsed_args): return 1 delete_nodes = [node] + failed = False for node in delete_nodes: if not node: continue @@ -1726,6 +1763,10 @@ def take_action(self, parsed_args): logger.error( f"Failed to delete node {node.name} ({node.id}): {exc}" ) + failed = True continue + + if failed: + return 1 finally: cleanup_cloud_environment(temp_files, original_cwd) diff --git a/osism/commands/redfish.py b/osism/commands/redfish.py index f70e0ad01..f9801bee4 100644 --- a/osism/commands/redfish.py +++ b/osism/commands/redfish.py @@ -169,9 +169,12 @@ def take_action(self, parsed_args): # Get column mappings for the resource type column_mappings = self._get_column_mappings(resourcetype) if column_mappings: - _, data_keys = self._get_filtered_columns( + headers, data_keys = self._get_filtered_columns( column_mappings, columns ) + if not headers: + print("No valid columns specified") + return filtered_result = self._filter_json_data(result, data_keys) print(json.dumps(filtered_result, indent=2)) else: diff --git a/tests/unit/commands/test_baremetal.py b/tests/unit/commands/test_baremetal.py index 45604597a..2fa447529 100644 --- a/tests/unit/commands/test_baremetal.py +++ b/tests/unit/commands/test_baremetal.py @@ -1,9 +1,15 @@ # SPDX-License-Identifier: Apache-2.0 -from unittest.mock import MagicMock, patch +import json +import os +import subprocess +from unittest.mock import MagicMock, call, patch +import openstack.exceptions import pytest +import yaml +from osism import settings from osism.commands import baremetal # Each of these command classes follows the identical pattern: when the @@ -216,3 +222,1485 @@ def test_delete_all_without_confirmation_returns_1(): cmd = baremetal.BaremetalDelete(MagicMock(), MagicMock()) parsed_args = cmd.get_parser("test").parse_args(["--all"]) assert cmd.take_action(parsed_args) == 1 + + +# --- Test doubles for the behavior tests below --- + + +class FakeNode: + """Ironic node double. + + The commands mix attribute access (``node.provision_state``), item access + (``node["maintenance"]``), membership tests (``"instance_info" in node``) + and ``node.get(...)``, which a plain MagicMock does not support well. + """ + + def __init__(self, **fields): + defaults = { + "id": "uuid-1", + "name": "node1", + "provision_state": "available", + "maintenance": False, + "instance_info": {"image_source": "image"}, + "extra": {}, + "target_raid_config": None, + "properties": {}, + "power_state": "power on", + } + defaults.update(fields) + self._fields = defaults + + def __getattr__(self, item): + fields = self.__dict__.get("_fields", {}) + if item in fields: + return fields[item] + raise AttributeError(item) + + def __getitem__(self, item): + return self._fields[item] + + def __contains__(self, item): + return item in self._fields + + def get(self, item, default=None): + return self._fields.get(item, default) + + +def _cloud_helpers(conn, success=True): + setup = MagicMock(return_value=("pw", ["tempfile"], "/cwd", success)) + getconn = MagicMock(return_value=conn) + cleanup = MagicMock() + return setup, getconn, cleanup + + +def _patch_cloud(setup, getconn, cleanup): + return patch( + "osism.tasks.openstack.get_cloud_helpers", + return_value=(setup, getconn, cleanup), + ) + + +# --- _apply_metalbox_vars --- + + +def test_apply_metalbox_vars_sets_entries_when_ip_found(): + play_vars = {} + with patch( + "osism.commands.baremetal._get_metalbox_primary_ip4", + return_value="192.168.30.1", + ): + baremetal._apply_metalbox_vars(play_vars, MagicMock()) + + assert play_vars["hosts_additional_entries"] == { + "metalbox.osism.xyz": "192.168.30.1" + } + assert play_vars["docker_insecure_registries"] == ["metalbox:5001"] + + +def test_apply_metalbox_vars_leaves_vars_untouched_without_ip(): + play_vars = {"existing": True} + with patch("osism.commands.baremetal._get_metalbox_primary_ip4", return_value=None): + baremetal._apply_metalbox_vars(play_vars, MagicMock()) + + assert play_vars == {"existing": True} + + +# --- Cloud setup failure guard (identical in every command of the module) --- + +SETUP_FAILURE_COMMANDS = [ + (baremetal.BaremetalList, []), + (baremetal.BaremetalDeploy, ["node1"]), + (baremetal.BaremetalDump, ["node1", "--ironic"]), + (baremetal.BaremetalUndeploy, ["node1"]), + (baremetal.BaremetalBurnIn, ["node1"]), + (baremetal.BaremetalClean, ["node1"]), + (baremetal.BaremetalProvide, ["node1"]), + (baremetal.BaremetalMaintenanceSet, ["node1"]), + (baremetal.BaremetalMaintenanceUnset, ["node1"]), + (baremetal.BaremetalPowerOn, ["node1"]), + (baremetal.BaremetalPowerOff, ["node1"]), + (baremetal.BaremetalDelete, ["node1"]), +] + + +@pytest.mark.parametrize("cls,args", SETUP_FAILURE_COMMANDS) +def test_setup_failure_returns_1_without_connection(cls, args): + cmd = cls(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + setup, getconn, cleanup = _cloud_helpers(MagicMock(), success=False) + with _patch_cloud(setup, getconn, cleanup): + assert cmd.take_action(parsed_args) == 1 + getconn.assert_not_called() + + +# --- BaremetalList --- + + +def _list_node(name, power_state="power on"): + return { + "name": name, + "id": f"uuid-{name}", + "power_state": power_state, + "provision_state": "active", + "maintenance": False, + } + + +def _run_list(args, conn, nb=None, info_mock=None): + cmd = baremetal.BaremetalList(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + setup, getconn, cleanup = _cloud_helpers(conn) + if info_mock is None: + info_mock = MagicMock(return_value={}) + with _patch_cloud(setup, getconn, cleanup), patch( + "osism.tasks.openstack.get_baremetal_node_netbox_info", info_mock + ), patch.dict("osism.utils.__dict__", {"nb": nb}): + rc = cmd.take_action(parsed_args) + return rc, info_mock + + +def test_list_rows_sorted_and_power_state_placeholder(capsys): + conn = MagicMock() + conn.baremetal.nodes.return_value = [ + _list_node("node-b", power_state=None), + _list_node("node-a"), + ] + + _run_list([], conn) + + out = capsys.readouterr().out + assert out.index("node-a") < out.index("node-b") + assert "n/a" in out + + +def test_list_query_flags_forwarded(): + conn = MagicMock() + conn.baremetal.nodes.return_value = [] + + _run_list(["--provision-state", "active", "--maintenance"], conn) + + conn.baremetal.nodes.assert_called_once_with( + provision_state="active", maintenance=True + ) + + +def test_list_without_flags_queries_without_kwargs(): + conn = MagicMock() + conn.baremetal.nodes.return_value = [] + + _run_list([], conn) + + conn.baremetal.nodes.assert_called_once_with() + + +def test_list_netbox_adds_device_role_column(capsys): + conn = MagicMock() + conn.baremetal.nodes.return_value = [ + _list_node("node-b"), + _list_node("node-a"), + ] + info_mock = MagicMock( + side_effect=[{"device_role": "compute"}, {"device_role": None}] + ) + + _run_list(["--netbox"], conn, nb=MagicMock(), info_mock=info_mock) + + out = capsys.readouterr().out + assert "Device Role" in out + assert "compute" in out + assert "N/A" in out + # Rows are sorted before the NetBox lookups run. + assert info_mock.call_args_list == [call("node-a"), call("node-b")] + + +def test_list_netbox_without_connection_skips_extra_column(capsys): + conn = MagicMock() + conn.baremetal.nodes.return_value = [_list_node("node-a")] + + _, info_mock = _run_list(["--netbox"], conn, nb=None) + + info_mock.assert_not_called() + assert "Device Role" not in capsys.readouterr().out + + +def test_list_cleanup_called_when_listing_fails(): + cmd = baremetal.BaremetalList(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args([]) + conn = MagicMock() + conn.baremetal.nodes.side_effect = RuntimeError("boom") + setup, getconn, cleanup = _cloud_helpers(conn) + + with _patch_cloud(setup, getconn, cleanup): + with pytest.raises(RuntimeError): + cmd.take_action(parsed_args) + + cleanup.assert_called_once_with(["tempfile"], "/cwd") + + +# --- BaremetalDeploy --- + + +def _run_deploy(args, conn, nb=None, pack_side_effect=None): + cmd = baremetal.BaremetalDeploy(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + setup, getconn, cleanup = _cloud_helpers(conn) + vault = MagicMock() + captured = {} + + def fake_pack(tmp_dir): + with open(os.path.join(tmp_dir, "playbook.yml")) as handle: + captured["playbook"] = yaml.safe_load(handle) + return "config-drive" + + with _patch_cloud(setup, getconn, cleanup), patch( + "osism.tasks.conductor.utils.get_vault", return_value=vault + ), patch("osism.tasks.conductor.utils.deep_decrypt") as deep_decrypt, patch( + "openstack.baremetal.configdrive.pack", + side_effect=pack_side_effect or fake_pack, + ), patch( + "osism.commands.baremetal._get_metalbox_primary_ip4", return_value=None + ), patch.dict( + "osism.utils.__dict__", {"nb": nb} + ): + rc = cmd.take_action(parsed_args) + return rc, captured, deep_decrypt, vault + + +@pytest.mark.parametrize( + "provision_state,maintenance,rebuild,expected_target", + [ + ("available", False, False, "active"), + ("deploy failed", False, False, "active"), + ("error", False, False, "rebuild"), + ("active", False, True, "rebuild"), + ("active", False, False, None), + ("available", True, False, None), + ], +) +def test_deploy_provision_state_decision( + provision_state, maintenance, rebuild, expected_target +): + node = FakeNode(provision_state=provision_state, maintenance=maintenance) + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + args = ["node1"] + (["--rebuild"] if rebuild else []) + _run_deploy(args, conn) + + if expected_target is None: + conn.baremetal.set_node_provision_state.assert_not_called() + else: + conn.baremetal.set_node_provision_state.assert_called_once_with( + node.id, + expected_target, + config_drive="config-drive", + deploy_steps=None, + ) + + +def test_deploy_refreshes_instance_info_from_extra(): + node = FakeNode( + instance_info={}, + extra={"instance_info": json.dumps({"image_source": "img"})}, + ) + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.update_node.side_effect = lambda n, **kwargs: n + + _run_deploy(["node1"], conn) + + conn.baremetal.update_node.assert_called_once_with( + node, instance_info={"image_source": "img"} + ) + + +def test_deploy_validation_failure_skips_node(): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.validate_node.side_effect = openstack.exceptions.ValidationException( + "invalid" + ) + + rc, _, _, _ = _run_deploy(["node1"], conn) + + assert rc == 1 + conn.baremetal.set_node_provision_state.assert_not_called() + + +def test_deploy_uses_netbox_local_context_data(): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + device = MagicMock() + device.local_context_data = { + "base_var": "value", + "frr_parameters": {"stale": True}, + "netplan_parameters": {"stale": True}, + } + nb = MagicMock() + nb.dcim.devices.get.return_value = device + + _, captured, _, _ = _run_deploy(["node1"], conn, nb=nb) + + play = captured["playbook"][0] + assert play["vars"]["base_var"] == "value" + assert play["vars"]["hostname_name"] == "node1" + assert "frr_parameters" not in play["vars"] + assert "netplan_parameters" not in play["vars"] + + +def test_deploy_finds_device_via_inventory_hostname(): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + device = MagicMock() + device.local_context_data = {"base_var": "value"} + nb = MagicMock() + nb.dcim.devices.get.return_value = None + nb.dcim.devices.filter.return_value = [device] + + _, captured, _, _ = _run_deploy(["node1"], conn, nb=nb) + + nb.dcim.devices.filter.assert_called_once_with(cf_inventory_hostname="node1") + assert captured["playbook"][0]["vars"]["base_var"] == "value" + + +def test_deploy_continues_when_netbox_lookup_fails(loguru_logs): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + nb = MagicMock() + nb.dcim.devices.get.side_effect = RuntimeError("netbox down") + + _, captured, _, _ = _run_deploy(["node1"], conn, nb=nb) + + conn.baremetal.set_node_provision_state.assert_called_once() + assert captured["playbook"][0]["vars"]["hostname_name"] == "node1" + assert any( + record["level"] == "WARNING" + and "Failed to fetch NetBox data" in record["message"] + for record in loguru_logs + ) + + +def test_deploy_netplan_parameters_extend_play(): + node = FakeNode( + extra={"netplan_parameters": json.dumps({"network_ethernets": {"eth0": {}}})} + ) + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _, captured, _, _ = _run_deploy(["node1"], conn) + + play = captured["playbook"][0] + assert play["vars"]["network_allow_service_restart"] is True + assert play["vars"]["network_ethernets"] == {"eth0": {}} + assert "osism.commons.network" in play["roles"] + + +def test_deploy_frr_parameters_extend_play(): + node = FakeNode(extra={"frr_parameters": json.dumps({"frr_local_as": 65000})}) + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _, captured, deep_decrypt, vault = _run_deploy(["node1"], conn) + + play = captured["playbook"][0] + assert play["vars"]["frr_dummy_interface"] == settings.FRR_DUMMY_INTERFACE + assert play["vars"]["frr_local_as"] == 65000 + assert "osism.services.frr" in play["roles"] + deep_decrypt.assert_called_once_with({"frr_local_as": 65000}, vault) + + +def test_deploy_supermicro_sets_cdrom_boot_device(): + node = FakeNode(properties={"vendor": " Supermicro "}) + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_deploy(["node1"], conn) + + conn.baremetal.set_node_boot_device.assert_called_once_with( + node.id, "cdrom", persistent=False + ) + + +def test_deploy_boot_device_failure_does_not_block_deploy(): + node = FakeNode(properties={"vendor": "Supermicro"}) + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.set_node_boot_device.side_effect = RuntimeError("ipmi error") + + _run_deploy(["node1"], conn) + + conn.baremetal.set_node_provision_state.assert_called_once() + + +def test_deploy_with_target_raid_config_passes_deploy_steps(): + raid_config = {"logical_disks": [{"size_gb": 100}]} + node = FakeNode(target_raid_config=raid_config) + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_deploy(["node1"], conn) + + kwargs = conn.baremetal.set_node_provision_state.call_args.kwargs + assert kwargs["deploy_steps"] == [ + { + "interface": "deploy", + "step": "erase_devices_metadata", + "args": {}, + "priority": 95, + }, + { + "interface": "raid", + "step": "apply_configuration", + "args": {"delete_existing": True, "raid_config": raid_config}, + "priority": 90, + }, + ] + + +def test_deploy_config_drive_failure_skips_node(loguru_logs): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + rc, _, _, _ = _run_deploy( + ["node1"], conn, pack_side_effect=RuntimeError("pack failed") + ) + + assert rc == 1 + conn.baremetal.set_node_provision_state.assert_not_called() + assert any( + "Failed to build config drive" in record["message"] for record in loguru_logs + ) + + +def test_deploy_all_continues_after_provision_failure(): + nodes = [ + FakeNode(name="node1", id="uuid-1"), + FakeNode(name="node2", id="uuid-2"), + ] + conn = MagicMock() + conn.baremetal.nodes.return_value = nodes + conn.baremetal.set_node_provision_state.side_effect = [ + RuntimeError("boom"), + None, + ] + + rc, _, _, _ = _run_deploy(["--all"], conn) + + # The failing node must not abort the loop, but the partially failed + # run must still report a non-zero exit code. + assert rc == 1 + conn.baremetal.nodes.assert_called_once_with(details=True) + conn.baremetal.find_node.assert_not_called() + assert conn.baremetal.set_node_provision_state.call_count == 2 + + +# --- BaremetalDump (happy paths) --- + + +def _run_dump(args, conn=None, nb=None, metalbox_side_effect=None): + cmd = baremetal.BaremetalDump(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + setup, getconn, cleanup = _cloud_helpers(conn or MagicMock()) + vault = MagicMock() + metalbox = MagicMock(return_value=None, side_effect=metalbox_side_effect) + with _patch_cloud(setup, getconn, cleanup), patch( + "osism.tasks.conductor.utils.get_vault", return_value=vault + ), patch("osism.tasks.conductor.utils.deep_decrypt") as deep_decrypt, patch( + "osism.commands.baremetal._get_metalbox_primary_ip4", metalbox + ), patch.dict( + "osism.utils.__dict__", {"nb": nb} + ): + rc = cmd.take_action(parsed_args) + return rc, deep_decrypt, vault + + +def test_dump_ironic_prints_playbook(capsys): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + rc, _, _ = _run_dump(["node1", "--ironic"], conn=conn) + + assert rc is None + play = yaml.safe_load(capsys.readouterr().out)[0] + assert play["vars"]["hostname_name"] == "node1" + assert play["roles"] == [ + "osism.commons.hostname", + "osism.commons.hosts", + "osism.commons.operator", + ] + assert play["tasks"][0]["ansible.builtin.systemd"] == { + "name": "rsyslog", + "state": "restarted", + } + + +def test_dump_ironic_extra_parameters(capsys): + node = FakeNode( + extra={ + "netplan_parameters": json.dumps({"network_ethernets": {"eth0": {}}}), + "frr_parameters": json.dumps({"frr_local_as": 65000}), + } + ) + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _, deep_decrypt, vault = _run_dump(["node1", "--ironic"], conn=conn) + + play = yaml.safe_load(capsys.readouterr().out)[0] + assert play["vars"]["network_allow_service_restart"] is True + assert play["vars"]["network_ethernets"] == {"eth0": {}} + assert play["vars"]["frr_local_as"] == 65000 + assert play["vars"]["frr_dummy_interface"] == settings.FRR_DUMMY_INTERFACE + assert "osism.commons.network" in play["roles"] + assert "osism.services.frr" in play["roles"] + deep_decrypt.assert_called_once_with({"frr_local_as": 65000}, vault) + + +def test_dump_ironic_generation_error_returns_1(capsys, loguru_logs): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + device = MagicMock() + device.name = "node1" + device.local_context_data = None + nb = MagicMock() + nb.dcim.devices.get.return_value = device + + rc, _, _ = _run_dump( + ["node1", "--ironic"], + conn=conn, + nb=nb, + metalbox_side_effect=RuntimeError("boom"), + ) + + assert rc == 1 + assert capsys.readouterr().out == "" + assert any( + record["level"] == "ERROR" + and "Failed to generate playbook" in record["message"] + for record in loguru_logs + ) + + +def test_dump_netbox_device_by_name(capsys): + device = MagicMock() + device.name = "node1" + device.local_context_data = { + "base_var": "value", + "frr_parameters": {"stale": True}, + "netplan_parameters": {"stale": True}, + } + device.custom_fields = { + "netplan_parameters": {"network_ethernets": {"eth0": {}}}, + "frr_parameters": {"frr_local_as": 65000}, + } + nb = MagicMock() + nb.dcim.devices.get.return_value = device + + rc, deep_decrypt, vault = _run_dump(["node1"], nb=nb) + + assert rc is None + play = yaml.safe_load(capsys.readouterr().out)[0] + assert play["vars"]["hostname_name"] == "node1" + assert play["vars"]["base_var"] == "value" + assert "stale" not in play["vars"] + assert play["vars"]["network_allow_service_restart"] is True + assert play["vars"]["network_ethernets"] == {"eth0": {}} + assert play["vars"]["frr_local_as"] == 65000 + assert play["vars"]["frr_dummy_interface"] == settings.FRR_DUMMY_INTERFACE + assert "osism.commons.network" in play["roles"] + assert "osism.services.frr" in play["roles"] + deep_decrypt.assert_called_once_with({"frr_local_as": 65000}, vault) + + +def test_dump_netbox_device_via_inventory_hostname(capsys): + device = MagicMock() + device.name = "node1" + device.local_context_data = None + device.custom_fields = {} + nb = MagicMock() + nb.dcim.devices.get.return_value = None + nb.dcim.devices.filter.return_value = [device] + + _run_dump(["node1"], nb=nb) + + nb.dcim.devices.filter.assert_called_once_with(cf_inventory_hostname="node1") + play = yaml.safe_load(capsys.readouterr().out)[0] + assert play["vars"]["hostname_name"] == "node1" + + +def test_dump_netbox_generation_error_returns_1(capsys, loguru_logs): + device = MagicMock() + device.name = "node1" + device.local_context_data = None + device.custom_fields = {} + nb = MagicMock() + nb.dcim.devices.get.return_value = device + + rc, _, _ = _run_dump(["node1"], nb=nb, metalbox_side_effect=RuntimeError("boom")) + + assert rc == 1 + assert capsys.readouterr().out == "" + assert any( + record["level"] == "ERROR" + and "Failed to generate playbook" in record["message"] + for record in loguru_logs + ) + + +# --- BaremetalUndeploy --- + + +def _run_undeploy(args, conn, ssh_cleanup_result=True): + cmd = baremetal.BaremetalUndeploy(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + setup, getconn, cleanup = _cloud_helpers(conn) + ssh_cleanup = MagicMock(return_value=ssh_cleanup_result) + with _patch_cloud(setup, getconn, cleanup), patch( + "osism.commands.baremetal.cleanup_ssh_known_hosts_for_node", ssh_cleanup + ): + rc = cmd.take_action(parsed_args) + return rc, ssh_cleanup + + +@pytest.mark.parametrize( + "state", ["active", "wait call-back", "deploy failed", "error"] +) +def test_undeploy_supported_states(state): + node = FakeNode(provision_state=state) + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.set_node_provision_state.return_value = node + + _, ssh_cleanup = _run_undeploy(["node1"], conn) + + conn.baremetal.set_node_provision_state.assert_called_once_with(node.id, "undeploy") + ssh_cleanup.assert_called_once_with("node1") + + +def test_undeploy_ssh_cleanup_success_logged(loguru_logs): + node = FakeNode(provision_state="active") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.set_node_provision_state.return_value = node + + _run_undeploy(["node1"], conn, ssh_cleanup_result=True) + + assert any( + record["level"] == "INFO" + and "SSH known_hosts cleanup completed successfully" in record["message"] + for record in loguru_logs + ) + + +def test_undeploy_ssh_cleanup_warning_logged(loguru_logs): + node = FakeNode(provision_state="active") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.set_node_provision_state.return_value = node + + _run_undeploy(["node1"], conn, ssh_cleanup_result=False) + + assert any( + record["level"] == "WARNING" + and "SSH known_hosts cleanup completed with warnings" in record["message"] + for record in loguru_logs + ) + + +def test_undeploy_unsupported_state_skipped(loguru_logs): + node = FakeNode(provision_state="available") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _, ssh_cleanup = _run_undeploy(["node1"], conn) + + conn.baremetal.set_node_provision_state.assert_not_called() + ssh_cleanup.assert_not_called() + assert any( + record["level"] == "WARNING" + and "not in supported provision state" in record["message"] + for record in loguru_logs + ) + + +def test_undeploy_provision_failure_returns_1(): + node = FakeNode(provision_state="active") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.set_node_provision_state.side_effect = RuntimeError("boom") + + rc, ssh_cleanup = _run_undeploy(["node1"], conn) + + assert rc == 1 + ssh_cleanup.assert_not_called() + + +def test_undeploy_all_iterates_nodes(): + nodes = [ + FakeNode(name="node1", id="uuid-1", provision_state="active"), + FakeNode(name="node2", id="uuid-2", provision_state="active"), + ] + conn = MagicMock() + conn.baremetal.nodes.return_value = nodes + conn.baremetal.set_node_provision_state.side_effect = lambda node_id, state: next( + n for n in nodes if n.id == node_id + ) + + _run_undeploy(["--all", "--yes-i-really-really-mean-it"], conn) + + conn.baremetal.nodes.assert_called_once_with() + assert conn.baremetal.set_node_provision_state.call_count == 2 + + +# --- BaremetalPing._ping_host --- + + +def _ping(run_result=None, side_effect=None): + cmd = baremetal.BaremetalPing(MagicMock(), MagicMock()) + results = {} + run_mock = MagicMock(return_value=run_result, side_effect=side_effect) + with patch("osism.commands.baremetal.subprocess.run", run_mock): + cmd._ping_host("10.0.0.1", results, "node1") + return results["node1"] + + +def _ping_output(returncode, stdout): + result = MagicMock() + result.returncode = returncode + result.stdout = stdout + return result + + +def test_ping_host_success_with_round_trip_line(): + stdout = ( + "PING 10.0.0.1 (10.0.0.1): 56 data bytes\n" + "3 packets transmitted, 3 packets received, 0% packet loss\n" + "round-trip min/avg/max = 1.0/2.0/3.0 ms\n" + ) + result = _ping(_ping_output(0, stdout)) + assert result == { + "host": "10.0.0.1", + "status": "SUCCESS", + "time_info": "1.0/2.0/3.0 ms", + } + + +def test_ping_host_partial_packet_loss(): + stdout = "3 packets transmitted, 2 packets received, 33% packet loss\n" + result = _ping(_ping_output(0, stdout)) + assert result["status"] == "PARTIAL (33% packet loss)" + assert result["time_info"] == "N/A" + + +def test_ping_host_success_without_stats_line(): + result = _ping(_ping_output(0, "PING 10.0.0.1 (10.0.0.1): 56 data bytes\n")) + assert result["status"] == "SUCCESS" + assert result["time_info"] == "N/A" + + +def test_ping_host_parses_linux_rtt_line(): + stdout = ( + "3 packets transmitted, 3 received, 0% packet loss, time 2003ms\n" + "rtt min/avg/max/mdev = 1.1/2.2/3.3/0.4 ms\n" + ) + result = _ping(_ping_output(0, stdout)) + assert result["status"] == "SUCCESS" + assert result["time_info"] == "1.1/2.2/3.3/0.4 ms" + + +def test_ping_host_nonzero_returncode_fails(): + result = _ping(_ping_output(1, "")) + assert result["status"] == "FAILED" + assert result["time_info"] == "N/A" + + +def test_ping_host_timeout_reports_error(): + error = subprocess.TimeoutExpired(cmd="ping", timeout=20) + result = _ping(side_effect=error) + assert result["status"] == "ERROR" + assert result["time_info"] == str(error)[:50] + + +# --- BaremetalPing.take_action (device discovery) --- + + +def _nb_device( + name, + ip="10.0.0.1/24", + power_state="power on", + provision_state="active", + has_ip=True, +): + device = MagicMock() + device.name = name + device.custom_fields = { + "power_state": power_state, + "provision_state": provision_state, + } + if has_ip: + device.primary_ip4.address = ip + else: + device.primary_ip4 = None + return device + + +def _run_ping_all(devices_per_query, queries=None, ping_impl=None): + cmd = baremetal.BaremetalPing(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args([]) + + if ping_impl is None: + + def ping_impl(self, host, results, host_name): + results[host_name] = { + "host": host, + "status": "SUCCESS", + "time_info": "1.0/2.0/3.0 ms", + } + + with patch.dict("osism.utils.__dict__", {"nb": MagicMock()}), patch( + "osism.tasks.conductor.netbox.get_nb_device_query_list_ironic", + return_value=queries or [{"role": "server"}], + ), patch( + "osism.tasks.netbox.get_devices", side_effect=devices_per_query + ) as get_devices, patch.object( + baremetal.BaremetalPing, "_ping_host", ping_impl + ): + rc = cmd.take_action(parsed_args) + return rc, get_devices + + +def test_ping_all_collects_devices_from_all_queries_and_filters(capsys): + matching = _nb_device("node-a") + wrong_power = _nb_device("node-b", power_state="power off") + wrong_state = _nb_device("node-c", provision_state="available") + + rc, get_devices = _run_ping_all( + devices_per_query=[[matching], [wrong_power, wrong_state]], + queries=[{"role": "server"}, {"role": "storage"}], + ) + + assert rc is None + assert get_devices.call_count == 2 + get_devices.assert_any_call(role="server") + get_devices.assert_any_call(role="storage") + out = capsys.readouterr().out + assert "node-a" in out + assert "node-b" not in out + assert "node-c" not in out + + +def test_ping_all_no_matching_devices_returns_none(loguru_logs): + rc, _ = _run_ping_all( + devices_per_query=[[_nb_device("node-b", power_state="power off")]] + ) + + assert rc is None + assert any( + "No devices found matching criteria" in record["message"] + for record in loguru_logs + ) + + +def test_ping_all_device_without_ip_excluded(capsys, loguru_logs): + with_ip = _nb_device("node-a") + without_ip = _nb_device("node-b", has_ip=False) + + _run_ping_all(devices_per_query=[[with_ip, without_ip]]) + + out = capsys.readouterr().out + assert "node-a" in out + assert "node-b" not in out + assert any( + record["level"] == "WARNING" and "no primary IPv4 address" in record["message"] + for record in loguru_logs + ) + + +def test_ping_all_only_ip_less_devices_returns_none(loguru_logs): + rc, _ = _run_ping_all(devices_per_query=[[_nb_device("node-a", has_ip=False)]]) + + assert rc is None + assert any( + "No devices found with primary IPv4 addresses" in record["message"] + for record in loguru_logs + ) + + +def test_ping_all_strips_prefix_from_ip(capsys): + _run_ping_all(devices_per_query=[[_nb_device("node-a", ip="10.0.0.5/24")]]) + + out = capsys.readouterr().out + assert "10.0.0.5" in out + assert "10.0.0.5/24" not in out + + +def test_ping_all_summary_counts_partial_as_failed(capsys): + def ping_impl(self, host, results, host_name): + status = "SUCCESS" if host_name == "node-a" else "PARTIAL (33% packet loss)" + results[host_name] = {"host": host, "status": status, "time_info": "N/A"} + + _run_ping_all( + devices_per_query=[ + [ + _nb_device("node-a", ip="10.0.0.1/24"), + _nb_device("node-b", ip="10.0.0.2/24"), + ] + ], + ping_impl=ping_impl, + ) + + out = capsys.readouterr().out + assert "Summary: 1 successful, 1 failed/partial out of 2 total" in out + + +# --- BaremetalBurnIn --- + +DEFAULT_BURNIN_STEPS = [ + {"step": "burnin_cpu", "interface": "deploy"}, + {"step": "burnin_memory", "interface": "deploy"}, + {"step": "burnin_disk", "interface": "deploy"}, +] + + +def _run_burnin(args, conn): + cmd = baremetal.BaremetalBurnIn(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + setup, getconn, cleanup = _cloud_helpers(conn) + with _patch_cloud(setup, getconn, cleanup): + return cmd.take_action(parsed_args) + + +def test_burnin_manageable_uses_default_steps(): + node = FakeNode(provision_state="manageable") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_burnin(["node1"], conn) + + conn.baremetal.set_node_provision_state.assert_called_once_with( + node.id, "clean", clean_steps=DEFAULT_BURNIN_STEPS + ) + + +def test_burnin_no_disk_removes_only_disk_step(): + node = FakeNode(provision_state="manageable") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_burnin(["node1", "--no-disk"], conn) + + conn.baremetal.set_node_provision_state.assert_called_once_with( + node.id, + "clean", + clean_steps=[ + {"step": "burnin_cpu", "interface": "deploy"}, + {"step": "burnin_memory", "interface": "deploy"}, + ], + ) + + +def test_burnin_available_node_moved_to_manageable_first(): + available = FakeNode(provision_state="available") + manageable = FakeNode(provision_state="manageable") + conn = MagicMock() + conn.baremetal.find_node.return_value = available + conn.baremetal.set_node_provision_state.return_value = manageable + conn.baremetal.wait_for_nodes_provision_state.return_value = [manageable] + + _run_burnin(["node1"], conn) + + assert conn.baremetal.set_node_provision_state.call_args_list == [ + call("uuid-1", "manage"), + call("uuid-1", "clean", clean_steps=DEFAULT_BURNIN_STEPS), + ] + conn.baremetal.wait_for_nodes_provision_state.assert_called_once_with( + ["uuid-1"], "manageable" + ) + + +def test_burnin_manage_failure_skips_node(): + available = FakeNode(provision_state="available") + conn = MagicMock() + conn.baremetal.find_node.return_value = available + conn.baremetal.set_node_provision_state.side_effect = RuntimeError("boom") + + rc = _run_burnin(["node1"], conn) + + assert rc == 1 + assert conn.baremetal.set_node_provision_state.call_count == 1 + conn.baremetal.wait_for_nodes_provision_state.assert_not_called() + + +def test_burnin_manageable_refreshes_instance_info_and_boot_device(): + node = FakeNode( + provision_state="manageable", + instance_info={}, + extra={"instance_info": json.dumps({"image_source": "img"})}, + properties={"vendor": "Supermicro"}, + ) + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.update_node.side_effect = lambda n, **kwargs: n + + _run_burnin(["node1"], conn) + + conn.baremetal.update_node.assert_called_once_with( + node, instance_info={"image_source": "img"} + ) + conn.baremetal.set_node_boot_device.assert_called_once_with( + node.id, "cdrom", persistent=False + ) + + +def test_burnin_clean_failure_returns_1(): + node = FakeNode(provision_state="manageable") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.set_node_provision_state.side_effect = RuntimeError("boom") + + rc = _run_burnin(["node1"], conn) + + assert rc == 1 + + +def test_burnin_active_without_confirmation_refused(loguru_logs): + node = FakeNode(provision_state="active") + node.set_provision_state = MagicMock() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_burnin(["node1"], conn) + + node.set_provision_state.assert_not_called() + assert any( + record["level"] == "ERROR" + and "yes-i-really-really-mean-it" in record["message"] + for record in loguru_logs + ) + + +def test_burnin_active_with_confirmation_uses_service_steps(loguru_logs): + node = FakeNode(provision_state="active") + node.set_provision_state = MagicMock() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_burnin(["node1", "--yes-i-really-really-mean-it"], conn) + + node.set_provision_state.assert_called_once_with( + conn.baremetal, + "service", + service_steps=[ + {"step": "burnin_cpu", "interface": "deploy"}, + {"step": "burnin_memory", "interface": "deploy"}, + ], + ) + assert any("Skipping disk burn-in" in record["message"] for record in loguru_logs) + + +def test_burnin_unsupported_state_warns(loguru_logs): + node = FakeNode(provision_state="enroll") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + rc = _run_burnin(["node1"], conn) + + assert rc is None + conn.baremetal.set_node_provision_state.assert_not_called() + assert any( + record["level"] == "WARNING" and "not in supported state" in record["message"] + for record in loguru_logs + ) + + +# --- BaremetalClean --- + +ERASE_DEVICES_STEP = {"interface": "deploy", "step": "erase_devices"} +RAID_DELETE_STEP = {"interface": "raid", "step": "delete_configuration"} + + +def _run_baremetal_clean(args, conn): + cmd = baremetal.BaremetalClean(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + setup, getconn, cleanup = _cloud_helpers(conn) + with _patch_cloud(setup, getconn, cleanup): + return cmd.take_action(parsed_args) + + +def test_clean_manageable_without_raid_interface(): + node = FakeNode(provision_state="manageable") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_baremetal_clean(["node1"], conn) + + conn.baremetal.set_node_provision_state.assert_called_once_with( + node.id, "clean", clean_steps=[ERASE_DEVICES_STEP] + ) + + +def test_clean_raid_interface_prepends_delete_configuration(): + node = FakeNode(provision_state="manageable", raid_interface="agent") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_baremetal_clean(["node1"], conn) + + conn.baremetal.set_node_provision_state.assert_called_once_with( + node.id, "clean", clean_steps=[RAID_DELETE_STEP, ERASE_DEVICES_STEP] + ) + + +def test_clean_available_node_moved_to_manageable_first(loguru_logs): + available = FakeNode(provision_state="available") + manageable = FakeNode(provision_state="manageable") + conn = MagicMock() + conn.baremetal.find_node.return_value = available + conn.baremetal.set_node_provision_state.return_value = manageable + conn.baremetal.wait_for_nodes_provision_state.return_value = [manageable] + + _run_baremetal_clean(["node1"], conn) + + assert conn.baremetal.set_node_provision_state.call_args_list == [ + call("uuid-1", "manage"), + call("uuid-1", "clean", clean_steps=[ERASE_DEVICES_STEP]), + ] + assert any( + "Successfully initiated clean" in record["message"] for record in loguru_logs + ) + + +def test_clean_manage_failure_skips_node(): + available = FakeNode(provision_state="available") + conn = MagicMock() + conn.baremetal.find_node.return_value = available + conn.baremetal.set_node_provision_state.side_effect = RuntimeError("boom") + + rc = _run_baremetal_clean(["node1"], conn) + + assert rc == 1 + assert conn.baremetal.set_node_provision_state.call_count == 1 + + +def test_clean_failure_returns_1(loguru_logs): + node = FakeNode(provision_state="manageable") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.set_node_provision_state.side_effect = RuntimeError("boom") + + rc = _run_baremetal_clean(["node1"], conn) + + assert rc == 1 + assert any( + record["level"] == "WARNING" and "Clean of node" in record["message"] + for record in loguru_logs + ) + + +def test_clean_unsupported_state_warns(loguru_logs): + node = FakeNode(provision_state="active") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_baremetal_clean(["node1"], conn) + + conn.baremetal.set_node_provision_state.assert_not_called() + assert any( + record["level"] == "WARNING" and "not in supported state" in record["message"] + for record in loguru_logs + ) + + +# --- BaremetalProvide / Maintenance / Power (happy paths) --- + + +def _run_simple(cls, args, conn): + cmd = cls(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + setup, getconn, cleanup = _cloud_helpers(conn) + with _patch_cloud(setup, getconn, cleanup): + return cmd.take_action(parsed_args) + + +def test_provide_manageable_node(): + node = FakeNode(provision_state="manageable") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_simple(baremetal.BaremetalProvide, ["node1"], conn) + + conn.baremetal.set_node_provision_state.assert_called_once_with(node.id, "provide") + + +def test_provide_maintenance_node_skipped(loguru_logs): + node = FakeNode(provision_state="manageable", maintenance=True) + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_simple(baremetal.BaremetalProvide, ["node1"], conn) + + conn.baremetal.set_node_provision_state.assert_not_called() + assert any( + record["level"] == "WARNING" and "not in supported state" in record["message"] + for record in loguru_logs + ) + + +def test_provide_failure_warns(loguru_logs): + node = FakeNode(provision_state="manageable") + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.set_node_provision_state.side_effect = RuntimeError("boom") + + rc = _run_simple(baremetal.BaremetalProvide, ["node1"], conn) + + assert rc == 1 + assert any(record["level"] == "WARNING" for record in loguru_logs) + + +def test_maintenance_set_with_reason(): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_simple(baremetal.BaremetalMaintenanceSet, ["node1", "--reason", "foo"], conn) + + conn.baremetal.set_node_maintenance.assert_called_once_with(node, reason="foo") + + +def test_maintenance_set_failure_logged(loguru_logs): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.set_node_maintenance.side_effect = RuntimeError("boom") + + rc = _run_simple(baremetal.BaremetalMaintenanceSet, ["node1"], conn) + + assert rc == 1 + assert any(record["level"] == "ERROR" for record in loguru_logs) + + +def test_maintenance_unset(): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_simple(baremetal.BaremetalMaintenanceUnset, ["node1"], conn) + + conn.baremetal.unset_node_maintenance.assert_called_once_with(node) + + +def test_maintenance_unset_failure_logged(loguru_logs): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.unset_node_maintenance.side_effect = RuntimeError("boom") + + rc = _run_simple(baremetal.BaremetalMaintenanceUnset, ["node1"], conn) + + assert rc == 1 + assert any(record["level"] == "ERROR" for record in loguru_logs) + + +def test_power_on(): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_simple(baremetal.BaremetalPowerOn, ["node1"], conn) + + conn.baremetal.set_node_power_state.assert_called_once_with(node.id, "power on") + + +@pytest.mark.parametrize( + "args,target", + [(["node1"], "power off"), (["node1", "--soft"], "soft power off")], +) +def test_power_off(args, target): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + + _run_simple(baremetal.BaremetalPowerOff, args, conn) + + conn.baremetal.set_node_power_state.assert_called_once_with(node.id, target) + + +def test_power_failure_logged(loguru_logs): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.set_node_power_state.side_effect = RuntimeError("boom") + + rc = _run_simple(baremetal.BaremetalPowerOn, ["node1"], conn) + + assert rc == 1 + assert any(record["level"] == "ERROR" for record in loguru_logs) + + +# --- BaremetalDelete --- + + +def _run_delete(args, conn, nb=None, secondary=None): + cmd = baremetal.BaremetalDelete(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + setup, getconn, cleanup = _cloud_helpers(conn) + with _patch_cloud(setup, getconn, cleanup), patch.dict( + "osism.utils.__dict__", + {"nb": nb, "secondary_nb_list": secondary if secondary is not None else []}, + ): + return cmd.take_action(parsed_args) + + +def test_delete_removes_ports_before_node(): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.ports.return_value = [MagicMock(id="port-1"), MagicMock(id="port-2")] + + _run_delete(["node1"], conn) + + conn.baremetal.ports.assert_called_once_with(node_uuid=node.id) + assert conn.baremetal.delete_port.call_args_list == [ + call("port-1", ignore_missing=True), + call("port-2", ignore_missing=True), + ] + conn.baremetal.delete_node.assert_called_once_with(node.id, ignore_missing=True) + ordered = [ + name + for name, _, _ in conn.baremetal.mock_calls + if name in ("delete_port", "delete_node") + ] + assert ordered == ["delete_port", "delete_port", "delete_node"] + + +def test_delete_port_failure_does_not_block_deletion(): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.ports.return_value = [MagicMock(id="port-1"), MagicMock(id="port-2")] + conn.baremetal.delete_port.side_effect = [RuntimeError("boom"), None] + + _run_delete(["node1"], conn) + + assert conn.baremetal.delete_port.call_count == 2 + conn.baremetal.delete_node.assert_called_once_with(node.id, ignore_missing=True) + + +def test_delete_clears_primary_netbox_states(): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.ports.return_value = [] + device = MagicMock() + device.custom_fields = {"provision_state": "active", "power_state": "power on"} + nb = MagicMock() + nb.dcim.devices.get.return_value = device + + _run_delete(["node1"], conn, nb=nb) + + assert device.custom_fields["provision_state"] is None + assert device.custom_fields["power_state"] is None + device.save.assert_called_once_with() + + +def test_delete_primary_device_not_found_warns(loguru_logs): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.ports.return_value = [] + nb = MagicMock() + nb.dcim.devices.get.return_value = None + + rc = _run_delete(["node1"], conn, nb=nb) + + assert rc is None + conn.baremetal.delete_node.assert_called_once() + assert any( + record["level"] == "WARNING" + and "not found in primary NetBox" in record["message"] + for record in loguru_logs + ) + + +def test_delete_primary_netbox_failure_continues_with_secondary(loguru_logs): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.ports.return_value = [] + nb = MagicMock() + nb.dcim.devices.get.side_effect = RuntimeError("netbox down") + secondary_device = MagicMock() + secondary_device.custom_fields = {"provision_state": "active"} + secondary = MagicMock() + secondary.base_url = "https://secondary" + secondary.dcim.devices.get.return_value = secondary_device + + rc = _run_delete(["node1"], conn, nb=nb, secondary=[secondary]) + + assert rc is None + secondary_device.save.assert_called_once_with() + assert any( + record["level"] == "WARNING" + and "Failed to clear NetBox states" in record["message"] + for record in loguru_logs + ) + + +def test_delete_secondary_netbox_failure_warns(loguru_logs): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.ports.return_value = [] + secondary = MagicMock() + secondary.base_url = "https://secondary" + secondary.dcim.devices.get.side_effect = RuntimeError("boom") + + rc = _run_delete(["node1"], conn, nb=None, secondary=[secondary]) + + assert rc is None + assert any( + record["level"] == "WARNING" and "https://secondary" in record["message"] + for record in loguru_logs + ) + + +def test_delete_without_netbox_still_deletes_node(): + node = FakeNode() + conn = MagicMock() + conn.baremetal.find_node.return_value = node + conn.baremetal.ports.return_value = [] + + rc = _run_delete(["node1"], conn, nb=None) + + assert rc is None + conn.baremetal.delete_node.assert_called_once_with(node.id, ignore_missing=True) + + +def test_delete_all_continues_after_node_failure(loguru_logs): + nodes = [ + FakeNode(name="node1", id="uuid-1"), + FakeNode(name="node2", id="uuid-2"), + ] + conn = MagicMock() + conn.baremetal.nodes.return_value = nodes + conn.baremetal.ports.return_value = [] + conn.baremetal.delete_node.side_effect = [RuntimeError("boom"), None] + + rc = _run_delete(["--all", "--yes-i-really-really-mean-it"], conn, nb=None) + + # The failing node must not abort the loop, but the partially failed + # run must still report a non-zero exit code. + assert rc == 1 + assert conn.baremetal.delete_node.call_count == 2 + assert any( + record["level"] == "ERROR" and "Failed to delete node" in record["message"] + for record in loguru_logs + ) diff --git a/tests/unit/commands/test_redfish.py b/tests/unit/commands/test_redfish.py new file mode 100644 index 000000000..b594999cc --- /dev/null +++ b/tests/unit/commands/test_redfish.py @@ -0,0 +1,298 @@ +# SPDX-License-Identifier: Apache-2.0 + +import json +from unittest.mock import MagicMock, patch + +import pytest + +from osism.commands import redfish + +from ._helpers import assert_not_called_before_lock_check + +ETHERNET_INTERFACES = [ + { + "id": "eth0", + "name": "Ethernet 0", + "description": "Onboard NIC", + "mac_address": "aa:bb:cc:dd:ee:00", + "permanent_mac_address": "aa:bb:cc:dd:ee:00", + "speed_mbps": 10000, + "mtu_size": 1500, + "link_status": "LinkUp", + "interface_enabled": True, + }, + { + "id": "eth1", + "name": "Ethernet 1", + "description": "Onboard NIC", + "mac_address": "aa:bb:cc:dd:ee:01", + "permanent_mac_address": "aa:bb:cc:dd:ee:01", + "speed_mbps": 25000, + "mtu_size": 9000, + "link_status": "LinkDown", + "interface_enabled": False, + }, +] + + +def _cmd(): + return redfish.List(MagicMock(), MagicMock()) + + +# --- _normalize_column_name --- + + +@pytest.mark.parametrize( + "raw,expected", + [ + ("MAC Address", "mac_address"), + ("mac_address", "mac_address"), + ("", ""), + (None, None), + ], +) +def test_normalize_column_name(raw, expected): + assert _cmd()._normalize_column_name(raw) == expected + + +# --- _get_column_mappings --- + + +def test_column_mappings_ethernet_interfaces(): + mappings = _cmd()._get_column_mappings("EthernetInterfaces") + assert mappings["ID"] == "id" + assert mappings["MAC Address"] == "mac_address" + assert mappings["Speed (Mbps)"] == "speed_mbps" + + +def test_column_mappings_network_adapters(): + mappings = _cmd()._get_column_mappings("NetworkAdapters") + assert mappings["Manufacturer"] == "manufacturer" + assert mappings["Firmware Version"] == "firmware_version" + + +def test_column_mappings_network_device_functions(): + mappings = _cmd()._get_column_mappings("NetworkDeviceFunctions") + assert mappings["Ethernet Enabled"] == "ethernet_enabled" + assert mappings["Adapter Name"] == "adapter_name" + + +def test_column_mappings_unknown_type_returns_none(): + assert _cmd()._get_column_mappings("Storage") is None + + +# --- _get_filtered_columns --- + + +def test_filtered_columns_defaults_to_all(): + cmd = _cmd() + mappings = cmd._get_column_mappings("EthernetInterfaces") + + headers, data_keys = cmd._get_filtered_columns(mappings, None) + assert headers == list(mappings.keys()) + assert data_keys == list(mappings.values()) + + headers, data_keys = cmd._get_filtered_columns(mappings, []) + assert headers == list(mappings.keys()) + assert data_keys == list(mappings.values()) + + +def test_filtered_columns_case_insensitive_selection_in_mapping_order(): + cmd = _cmd() + mappings = cmd._get_column_mappings("EthernetInterfaces") + + headers, data_keys = cmd._get_filtered_columns(mappings, ["mac address", "ID"]) + assert headers == ["ID", "MAC Address"] + assert data_keys == ["id", "mac_address"] + + +def test_filtered_columns_unknown_column_warns_but_keeps_valid(loguru_logs): + cmd = _cmd() + mappings = cmd._get_column_mappings("EthernetInterfaces") + + headers, data_keys = cmd._get_filtered_columns(mappings, ["bogus", "id"]) + assert headers == ["ID"] + assert data_keys == ["id"] + assert any( + record["level"] == "WARNING" + and "bogus" in record["message"] + and "Available columns" in record["message"] + for record in loguru_logs + ) + + +def test_filtered_columns_nothing_matches(): + cmd = _cmd() + mappings = cmd._get_column_mappings("EthernetInterfaces") + assert cmd._get_filtered_columns(mappings, ["bogus"]) == ([], []) + + +# --- _filter_json_data --- + + +def test_filter_json_data_reduces_to_selected_keys(): + data = [{"id": "eth0", "name": "Ethernet 0"}, {"id": "eth1"}] + filtered = _cmd()._filter_json_data(data, ["id", "name"]) + assert filtered == [ + {"id": "eth0", "name": "Ethernet 0"}, + {"id": "eth1", "name": None}, + ] + + +def test_filter_json_data_passthrough_on_empty_input(): + cmd = _cmd() + assert cmd._filter_json_data([], ["id"]) == [] + data = [{"id": "eth0"}] + assert cmd._filter_json_data(data, []) is data + + +# --- _filter_and_display_table --- + + +def test_filter_and_display_table_empty_data_prints_nothing(capsys): + cmd = _cmd() + mappings = cmd._get_column_mappings("EthernetInterfaces") + cmd._filter_and_display_table([], mappings) + assert capsys.readouterr().out == "" + + +def test_filter_and_display_table_all_columns_invalid(capsys): + cmd = _cmd() + mappings = cmd._get_column_mappings("EthernetInterfaces") + cmd._filter_and_display_table(ETHERNET_INTERFACES, mappings, ["bogus"]) + assert "No valid columns specified" in capsys.readouterr().out + + +def test_filter_and_display_table_happy_path(capsys): + cmd = _cmd() + mappings = cmd._get_column_mappings("EthernetInterfaces") + cmd._filter_and_display_table(ETHERNET_INTERFACES, mappings) + + out = capsys.readouterr().out + assert "MAC Address" in out + assert "aa:bb:cc:dd:ee:00" in out + assert "Total items: 2" in out + + +# --- take_action --- + + +def _run(args, result, lock_mock=None): + cmd = _cmd() + parsed_args = cmd.get_parser("test").parse_args(args) + + task = MagicMock() + task.delay.return_value.get.return_value = result + if lock_mock is None: + lock_mock = MagicMock() + with patch( + "osism.commands.redfish.utils.check_task_lock_and_exit", lock_mock + ), patch("osism.tasks.conductor.get_redfish_resources", task): + rc = cmd.take_action(parsed_args) + return rc, task + + +def test_take_action_checks_task_lock_before_dispatch(): + cmd = _cmd() + parsed_args = cmd.get_parser("test").parse_args(["host1", "EthernetInterfaces"]) + + task = MagicMock() + task.delay.return_value.get.return_value = [] + lock_mock = MagicMock(side_effect=assert_not_called_before_lock_check(task.delay)) + with patch( + "osism.commands.redfish.utils.check_task_lock_and_exit", lock_mock + ), patch("osism.tasks.conductor.get_redfish_resources", task): + cmd.take_action(parsed_args) + + lock_mock.assert_called_once_with() + task.delay.assert_called_once_with("host1", "EthernetInterfaces") + + +def test_take_action_json_full_dump(capsys): + _run(["host1", "EthernetInterfaces", "--format", "json"], ETHERNET_INTERFACES) + out = capsys.readouterr().out + assert out.strip() == json.dumps(ETHERNET_INTERFACES, indent=2) + + +def test_take_action_json_with_columns(capsys): + _run( + [ + "host1", + "EthernetInterfaces", + "--format", + "json", + "--column", + "mac address", + ], + ETHERNET_INTERFACES, + ) + out = capsys.readouterr().out + assert json.loads(out) == [ + {"mac_address": "aa:bb:cc:dd:ee:00"}, + {"mac_address": "aa:bb:cc:dd:ee:01"}, + ] + + +def test_take_action_json_all_invalid_columns_prints_message(capsys, loguru_logs): + # Consistent with the table path: an all-invalid column selection must + # not silently fall back to dumping every field. + _run( + ["host1", "EthernetInterfaces", "--format", "json", "--column", "bogus"], + ETHERNET_INTERFACES, + ) + out = capsys.readouterr().out + assert "No valid columns specified" in out + assert "aa:bb:cc:dd:ee:00" not in out + assert any( + record["level"] == "WARNING" + and "bogus" in record["message"] + and "Available columns" in record["message"] + for record in loguru_logs + ) + + +def test_take_action_json_columns_unknown_type_dumps_everything(capsys): + result = [{"id": "x", "capacity": 42}] + _run(["host1", "Storage", "--format", "json", "--column", "id"], result) + out = capsys.readouterr().out + assert json.loads(out) == result + + +def test_take_action_json_empty_result(capsys): + _run(["host1", "EthernetInterfaces", "--format", "json"], []) + assert capsys.readouterr().out.strip() == "[]" + + +def test_take_action_table_ethernet_interfaces(capsys): + _run(["host1", "EthernetInterfaces"], ETHERNET_INTERFACES) + out = capsys.readouterr().out + assert "MAC Address" in out + assert "Total items: 2" in out + + +def test_take_action_table_network_adapters(capsys): + adapters = [{"id": "nic1", "manufacturer": "ACME", "model": "X540"}] + _run(["host1", "NetworkAdapters"], adapters) + out = capsys.readouterr().out + assert "Manufacturer" in out + assert "ACME" in out + + +def test_take_action_table_network_device_functions(capsys): + functions = [{"id": "fn1", "adapter_name": "Adapter 1"}] + _run(["host1", "NetworkDeviceFunctions"], functions) + out = capsys.readouterr().out + assert "Adapter Name" in out + assert "Adapter 1" in out + + +def test_take_action_table_unknown_type_logs_result(capsys, loguru_logs): + _run(["host1", "Storage"], [{"id": "x"}]) + assert capsys.readouterr().out == "" + assert any("Retrieved resources" in record["message"] for record in loguru_logs) + + +def test_take_action_table_no_result(capsys): + _run(["host1", "EthernetInterfaces"], None) + out = capsys.readouterr().out + assert "No EthernetInterfaces resources found for host1" in out diff --git a/tests/unit/commands/test_server.py b/tests/unit/commands/test_server.py index edbe0eccd..99909d80a 100644 --- a/tests/unit/commands/test_server.py +++ b/tests/unit/commands/test_server.py @@ -1,8 +1,21 @@ +from datetime import datetime, timedelta, timezone from unittest.mock import MagicMock, patch from osism.commands import server +def _obj(**attrs): + """Build a MagicMock with real attribute values (incl. ``name``).""" + obj = MagicMock() + for key, value in attrs.items(): + setattr(obj, key, value) + return obj + + +def _created_at(seconds_ago): + return (datetime.now(timezone.utc) - timedelta(seconds=seconds_ago)).isoformat() + + def _run(args, conn): cmd = server.ServerList(MagicMock(), MagicMock()) parsed_args = cmd.get_parser("test").parse_args(args) @@ -71,3 +84,350 @@ def test_migrate_returns_1_when_server_not_active_or_paused(): ) result = _run_migrate(["someinstance"], conn) assert result == 1 + + +def _run_migrate_interactive(args, conn, prompt_return="yes"): + """Run ServerMigrate with prompt and time.sleep mocked out.""" + cmd = server.ServerMigrate(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + setup = MagicMock(return_value=("pw", [], None, True)) + getconn = MagicMock(return_value=conn) + cleanup = MagicMock() + prompt_mock = MagicMock(return_value=prompt_return) + sleep_mock = MagicMock() + with patch( + "osism.tasks.openstack.get_cloud_helpers", + return_value=(setup, getconn, cleanup), + ), patch("osism.commands.server.prompt", prompt_mock), patch( + "osism.commands.server.time.sleep", sleep_mock + ): + rc = cmd.take_action(parsed_args) + return rc, prompt_mock, sleep_mock + + +def test_migrate_active_with_yes_uses_defaults(): + conn = MagicMock() + active = _obj(id="s1", name="srv1", status="ACTIVE") + conn.compute.get_server.side_effect = [active, active] + + _, prompt_mock, _ = _run_migrate_interactive(["--yes", "s1"], conn) + + conn.compute.live_migrate_server.assert_called_once_with( + "s1", host=None, block_migration="auto", force=False + ) + prompt_mock.assert_not_called() + + +def test_migrate_target_and_force_passed_through(): + conn = MagicMock() + active = _obj(id="s1", name="srv1", status="ACTIVE") + conn.compute.get_server.side_effect = [active, active] + + _run_migrate_interactive(["--yes", "--target", "host1", "--force", "s1"], conn) + + conn.compute.live_migrate_server.assert_called_once_with( + "s1", host="host1", block_migration="auto", force=True + ) + + +def test_migrate_prompt_no_skips_migration(): + conn = MagicMock() + active = _obj(id="s1", name="srv1", status="ACTIVE") + conn.compute.get_server.return_value = active + + _, prompt_mock, _ = _run_migrate_interactive(["s1"], conn, prompt_return="no") + + prompt_mock.assert_called_once() + conn.compute.live_migrate_server.assert_not_called() + + +def test_migrate_prompt_y_accepted(): + conn = MagicMock() + active = _obj(id="s1", name="srv1", status="ACTIVE") + conn.compute.get_server.side_effect = [active, active] + + _run_migrate_interactive(["s1"], conn, prompt_return="y") + + conn.compute.live_migrate_server.assert_called_once() + + +def test_migrate_waits_until_no_longer_migrating(): + conn = MagicMock() + active = _obj(id="s1", name="srv1", status="ACTIVE") + migrating = _obj(id="s1", name="srv1", status="MIGRATING") + conn.compute.get_server.side_effect = [active, migrating, migrating, active] + + _, _, sleep_mock = _run_migrate_interactive(["--yes", "s1"], conn) + + assert conn.compute.get_server.call_count == 4 + assert sleep_mock.call_count == 3 + + +def test_migrate_no_wait_skips_polling(): + conn = MagicMock() + active = _obj(id="s1", name="srv1", status="ACTIVE") + conn.compute.get_server.side_effect = [active] + + _, _, sleep_mock = _run_migrate_interactive(["--yes", "--no-wait", "s1"], conn) + + assert conn.compute.get_server.call_count == 1 + sleep_mock.assert_not_called() + conn.compute.live_migrate_server.assert_called_once() + + +def test_migrate_paused_server_allowed(): + conn = MagicMock() + paused = _obj(id="s1", name="srv1", status="PAUSED") + conn.compute.get_server.side_effect = [paused] + + rc, _, _ = _run_migrate_interactive(["--yes", "--no-wait", "s1"], conn) + + assert rc is None + conn.compute.live_migrate_server.assert_called_once() + + +# --- ServerList happy paths --- + + +def test_list_domain_happy_path(capsys): + conn = MagicMock() + domain = _obj(id="d1", name="dom1") + conn.identity.find_domain.return_value = domain + project = _obj(id="p1", name="proj1") + conn.identity.projects.return_value = [project] + srv = _obj( + id="s1", + name="vm1", + status="ACTIVE", + user_id="u1", + flavor={"original_name": "m1.small"}, + ) + conn.compute.servers.return_value = [srv] + + _run(["--domain", "dom1"], conn) + + conn.identity.projects.assert_called_once_with(domain_id="d1") + conn.compute.servers.assert_called_once_with(all_projects=True, project_id="p1") + out = capsys.readouterr().out + for header in ["Project", "Project ID", "User ID", "ID", "Name", "Flavor"]: + assert header in out + assert "proj1" in out + assert "vm1" in out + assert "m1.small" in out + + +def test_list_project_with_project_domain(capsys): + conn = MagicMock() + project_domain = _obj(id="pd1") + conn.identity.find_domain.return_value = project_domain + project = _obj(id="p1", name="proj1", domain_id="pd1") + conn.identity.find_project.return_value = project + resolved_domain = _obj(name="domname") + conn.identity.get_domain.return_value = resolved_domain + srv = _obj( + id="s1", + name="vm1", + status="ACTIVE", + user_id="u1", + flavor={"original_name": "f1"}, + ) + conn.compute.servers.return_value = [srv] + + _run(["--project", "proj1", "--project-domain", "pd"], conn) + + conn.identity.find_project.assert_called_once_with("proj1", domain_id="pd1") + out = capsys.readouterr().out + assert "domname" in out + assert "vm1" in out + + +def test_list_project_get_domain_failure_falls_back_to_id(capsys): + conn = MagicMock() + project = _obj(id="p1", name="proj1", domain_id="pd1") + conn.identity.find_project.return_value = project + conn.identity.get_domain.side_effect = RuntimeError("keystone down") + srv = _obj( + id="s1", + name="vm1", + status="ACTIVE", + user_id="u1", + flavor={"original_name": "f1"}, + ) + conn.compute.servers.return_value = [srv] + + _run(["--project", "proj1"], conn) + + out = capsys.readouterr().out + assert "pd1" in out + + +def test_list_user_happy_path(capsys): + conn = MagicMock() + user = MagicMock() + user.id = "u1" + user.__contains__.return_value = True + conn.identity.find_user.return_value = user + srv = _obj( + id="s1", + name="vm1", + status="ACTIVE", + project_id="p1", + flavor={"original_name": "f1"}, + ) + conn.compute.servers.return_value = [srv] + project = _obj(id="p1", domain_id="d1") + conn.identity.get_project.return_value = project + domain = _obj(name="domname") + conn.identity.get_domain.return_value = domain + + _run(["--user", "alice"], conn) + + conn.compute.servers.assert_called_once_with(all_projects=True, user_id="u1") + out = capsys.readouterr().out + assert "domname" in out + assert "vm1" in out + assert "p1" in out + + +def test_list_user_domain_resolution_failure_leaves_domain_empty(capsys): + conn = MagicMock() + user = MagicMock() + user.id = "u1" + user.__contains__.return_value = True + conn.identity.find_user.return_value = user + srv = _obj( + id="s1", + name="vm1", + status="ACTIVE", + project_id="p1", + flavor={"original_name": "f1"}, + ) + conn.compute.servers.return_value = [srv] + conn.identity.get_project.side_effect = RuntimeError("keystone down") + + _run(["--user", "alice"], conn) + + out = capsys.readouterr().out + assert "vm1" in out + assert "domname" not in out + + +def test_list_default_only_reports_old_build_and_error_servers(capsys): + conn = MagicMock() + old_build = _obj( + id="old-build-id", + name="ob", + status="BUILD", + flavor={"original_name": "f"}, + created_at=_created_at(8000), + ) + fresh_build = _obj( + id="fresh-build-id", + name="fb", + status="BUILD", + flavor={"original_name": "f"}, + created_at=_created_at(60), + ) + old_error = _obj( + id="old-error-id", + name="oe", + status="ERROR", + flavor={"original_name": "f"}, + created_at=_created_at(8000), + ) + + def servers(all_projects=True, status=None, **kwargs): + return {"build": [old_build, fresh_build], "error": [old_error]}[status] + + conn.compute.servers.side_effect = servers + + _run([], conn) + + out = capsys.readouterr().out + assert "old-build-id" in out + assert "old-error-id" in out + assert "fresh-build-id" not in out + + +# --- ServerClean --- + + +def _run_clean(args, conn, prompt_return="yes"): + cmd = server.ServerClean(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + setup = MagicMock(return_value=("pw", [], None, True)) + getconn = MagicMock(return_value=conn) + cleanup = MagicMock() + prompt_mock = MagicMock(return_value=prompt_return) + with patch( + "osism.tasks.openstack.get_cloud_helpers", + return_value=(setup, getconn, cleanup), + ), patch("osism.commands.server.prompt", prompt_mock): + rc = cmd.take_action(parsed_args) + return rc, prompt_mock, cleanup + + +def _clean_conn(build=None, error=None): + conn = MagicMock() + + def servers(all_projects=True, status=None, **kwargs): + return {"build": build or [], "error": error or []}[status] + + conn.compute.servers.side_effect = servers + return conn + + +def test_clean_deletes_old_build_server_with_yes(): + stuck = _obj(id="s1", name="vm1", created_at=_created_at(8000)) + conn = _clean_conn(build=[stuck]) + + _, prompt_mock, _ = _run_clean(["--yes"], conn) + + conn.compute.delete_server.assert_called_once_with("s1", force=True) + prompt_mock.assert_not_called() + + +def test_clean_skips_young_build_server(): + fresh = _obj(id="s1", name="vm1", created_at=_created_at(60)) + conn = _clean_conn(build=[fresh]) + + _, prompt_mock, _ = _run_clean(["--yes"], conn) + + conn.compute.delete_server.assert_not_called() + prompt_mock.assert_not_called() + + +def test_clean_honors_custom_build_timeout(): + stuck = _obj(id="s1", name="vm1", created_at=_created_at(120)) + conn = _clean_conn(build=[stuck]) + + _run_clean(["--yes", "--build-timeout", "60"], conn) + + conn.compute.delete_server.assert_called_once_with("s1", force=True) + + +def test_clean_prompt_no_skips_deletion(): + stuck = _obj(id="s1", name="vm1", created_at=_created_at(8000)) + conn = _clean_conn(build=[stuck]) + + _, prompt_mock, _ = _run_clean([], conn, prompt_return="no") + + prompt_mock.assert_called_once() + conn.compute.delete_server.assert_not_called() + + +def test_clean_error_server_deleted_regardless_of_age(): + broken = _obj(id="s2", name="vm2", created_at=_created_at(60)) + conn = _clean_conn(error=[broken]) + + _run_clean(["--yes"], conn) + + conn.compute.delete_server.assert_called_once_with("s2", force=True) + + +def test_clean_cleanup_called_in_finally(): + conn = _clean_conn() + + _, _, cleanup = _run_clean(["--yes"], conn) + + cleanup.assert_called_once_with([], None) diff --git a/tests/unit/commands/test_stress.py b/tests/unit/commands/test_stress.py new file mode 100644 index 000000000..d97861ce3 --- /dev/null +++ b/tests/unit/commands/test_stress.py @@ -0,0 +1,124 @@ +# SPDX-License-Identifier: Apache-2.0 + +from unittest.mock import MagicMock, patch + +import pytest + +from osism.commands import stress + +STRESS_TOOL = "/openstack-simple-stress/openstack_simple_stress/main.py" + +BOOLEAN_FLAGS = [ + "--no-cleanup", + "--debug", + "--no-delete", + "--no-volume", + "--no-boot-volume", + "--no-wait", + "--clean", +] + + +def _run(args, run_mock=None, setup_success=True): + """Drive OpenStackStress.take_action with mocked cloud helpers.""" + cmd = stress.OpenStackStress(MagicMock(), MagicMock()) + parsed_args = cmd.get_parser("test").parse_args(args) + + setup = MagicMock(return_value=("pw", ["tempfile"], "/cwd", setup_success)) + cleanup = MagicMock() + if run_mock is None: + run_mock = MagicMock(return_value=MagicMock(returncode=0)) + with patch( + "osism.tasks.openstack.get_cloud_helpers", + return_value=(setup, MagicMock(), cleanup), + ), patch("osism.commands.stress.subprocess.run", run_mock): + result = cmd.take_action(parsed_args) + return result, run_mock, cleanup + + +def _flag_value(command, flag): + return command[command.index(flag) + 1] + + +def test_defaults_build_expected_command(): + result, run_mock, _ = _run([]) + + command = run_mock.call_args[0][0] + assert command[:2] == ["python3", STRESS_TOOL] + for flag in BOOLEAN_FLAGS: + assert flag not in command + + expected = { + "--interval": "10", + "--number": "1", + "--parallel": "1", + "--timeout": "600", + "--volume-number": "1", + "--volume-size": "1", + "--boot-volume-size": "20", + "--cloud": "simple-stress", + "--flavor": "SCS-1V-2", + "--image": "Ubuntu 24.04", + "--subnet-cidr": "10.100.0.0/16", + "--prefix": "simple-stress", + "--compute-zone": "nova", + "--storage-zone": "nova", + "--affinity": "soft-anti-affinity", + "--volume-type": "__DEFAULT__", + "--mode": "rolling", + } + for flag, value in expected.items(): + assert _flag_value(command, flag) == value + + assert result == 0 + + +@pytest.mark.parametrize("flag", BOOLEAN_FLAGS) +def test_boolean_flag_appended(flag): + _, run_mock, _ = _run([flag]) + assert flag in run_mock.call_args[0][0] + + +def test_custom_values_propagated(): + _, run_mock, _ = _run(["--number", "5", "--flavor", "X", "--volume-size", "10"]) + + command = run_mock.call_args[0][0] + assert _flag_value(command, "--number") == "5" + assert _flag_value(command, "--flavor") == "X" + assert _flag_value(command, "--volume-size") == "10" + + +@pytest.mark.parametrize("returncode", [0, 3]) +def test_returncode_passed_through(returncode): + run_mock = MagicMock(return_value=MagicMock(returncode=returncode)) + result, _, cleanup = _run([], run_mock=run_mock) + + assert result == returncode + cleanup.assert_called_once_with(["tempfile"], "/cwd") + + +def test_tool_not_found_returns_1(loguru_logs): + run_mock = MagicMock(side_effect=FileNotFoundError()) + result, _, cleanup = _run([], run_mock=run_mock) + + assert result == 1 + assert any( + record["level"] == "ERROR" and STRESS_TOOL in record["message"] + for record in loguru_logs + ) + cleanup.assert_called_once_with(["tempfile"], "/cwd") + + +def test_generic_exception_returns_1(): + run_mock = MagicMock(side_effect=RuntimeError("boom")) + result, _, cleanup = _run([], run_mock=run_mock) + + assert result == 1 + cleanup.assert_called_once_with(["tempfile"], "/cwd") + + +def test_setup_failure_returns_1(): + result, run_mock, _ = _run([], setup_success=False) + + assert result == 1 + run_mock.assert_not_called()