From df9d55f78d09f7ae33bebeb6f31b57ce0d686a10 Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Mon, 27 Apr 2026 22:59:42 +0200 Subject: [PATCH] Add unit tests for osism/tasks/conductor/{config,netbox} MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Covers the conductor's setup/lookup layer: - get_configuration (config.py): empty file → {} with warning; enable_ironic falsy/truthy strings parametrised; missing ironic_parameters logs error and returns config as-is. The image_source, deploy_kernel and deploy_ramdisk fields share one parametrised set of tests for name → UUID resolution, UUID and URL pass-through, and unresolved-keeps-original-with-warning; cleaning_network and provisioning_network share a parametrised pair against network_get; section-present-without-key no-op, empty ironic_parameters, full end-to-end resolution across all five fields, and the /etc/conductor.yml path are also verified. - get_nb_device_query_list_ironic and get_nb_device_query_list_sonic (netbox.py): the two helpers are structurally identical and now share one parametrised body covering all supported pass-through filters (site, region, site_group, tag, status), location and rack name → ID resolution with key rename, combined location+rack, empty list, YAMLError, non-list payload, string and null elements, unsupported filter key, and unresolved location/rack all raising the expected exception. Error tests assert on the actual logged message (\"Setting needs to be an array of mappings\" and \"Unknown value in \") instead of just call presence. - get_device_oob_ip (netbox.py): oob_ip field used directly without NetBox calls; fallback to mgmt-only interface lookup; non-mgmt interfaces skipped; first mgmt without IPs falls through to second; no mgmt interfaces, no interfaces at all, mgmt without IPs, and blank address strings all return None; IPv6 with prefix; malformed address, interfaces.filter raising and ip_addresses.filter raising each return None and assert on the \"Could not get OOB IP for device \" warning message; verifies keyword-argument shape (device_id and assigned_object_id); device without an oob_ip attribute falls back; first assigned address wins on multiple IPs. A shared loguru_logs fixture in tests/conftest.py installs a temporary loguru sink so tests can assert on actual log message content (pytest's built-in caplog only captures stdlib logging records, not loguru). Closes #2206 AI-assisted: Claude Code Signed-off-by: Christian Berendt --- tests/conftest.py | 28 ++ tests/unit/tasks/conductor/test_config.py | 352 +++++++++++++++++ tests/unit/tasks/conductor/test_netbox.py | 444 ++++++++++++++++++++++ 3 files changed, 824 insertions(+) create mode 100644 tests/unit/tasks/conductor/test_config.py create mode 100644 tests/unit/tasks/conductor/test_netbox.py diff --git a/tests/conftest.py b/tests/conftest.py index 7d0af13f6..5483f86b9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -17,6 +17,8 @@ import sys import types +import pytest + def _install_ansible_stubs() -> None: try: @@ -78,3 +80,29 @@ def __init__(self, *args, **kwargs): _install_ansible_stubs() + + +@pytest.fixture +def loguru_logs(): + """Capture loguru log records emitted during a test. + + Yields a list of ``{"level": str, "message": str}`` dicts. The default + pytest ``caplog`` fixture only sees ``logging`` records; loguru goes + elsewhere, so we install our own sink for the duration of the test. + """ + from loguru import logger + + records: list[dict[str, str]] = [] + handler_id = logger.add( + lambda message: records.append( + { + "level": message.record["level"].name, + "message": message.record["message"], + } + ), + level="DEBUG", + ) + try: + yield records + finally: + logger.remove(handler_id) diff --git a/tests/unit/tasks/conductor/test_config.py b/tests/unit/tasks/conductor/test_config.py new file mode 100644 index 000000000..aed594ab1 --- /dev/null +++ b/tests/unit/tasks/conductor/test_config.py @@ -0,0 +1,352 @@ +# SPDX-License-Identifier: Apache-2.0 + +from types import SimpleNamespace +from unittest.mock import mock_open + +import pytest +import yaml + +from osism.tasks.conductor import config as config_module +from osism.tasks.conductor.config import get_configuration + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +# (section, key) tuples for fields that resolve via ``openstack.image_get``. +IMAGE_FIELDS = [ + pytest.param("instance_info", "image_source", id="image_source"), + pytest.param("driver_info", "deploy_kernel", id="deploy_kernel"), + pytest.param("driver_info", "deploy_ramdisk", id="deploy_ramdisk"), +] + +# Keys under ``driver_info`` that resolve via ``openstack.network_get``. +NETWORK_FIELDS = [ + pytest.param("cleaning_network", id="cleaning_network"), + pytest.param("provisioning_network", id="provisioning_network"), +] + + +def _patch_open(mocker, payload): + """Route ``open`` inside ``config`` to an in-memory YAML payload.""" + if isinstance(payload, dict): + data = yaml.safe_dump(payload) + else: + data = payload + return mocker.patch( + "osism.tasks.conductor.config.open", mock_open(read_data=data), create=True + ) + + +def _payload_with(section, key, value): + """Build a minimal ``ironic_parameters`` payload with one field set.""" + return {"ironic_parameters": {section: {key: value}}} + + +def _has_log(records, level, substring): + return any(r["level"] == level and substring in r["message"] for r in records) + + +@pytest.fixture +def patch_openstack(mocker): + """Patch the openstack helpers imported into config.""" + image_get = mocker.patch("osism.tasks.conductor.config.openstack.image_get") + network_get = mocker.patch("osism.tasks.conductor.config.openstack.network_get") + return SimpleNamespace(image_get=image_get, network_get=network_get) + + +@pytest.fixture +def enable_ironic(mocker): + """Toggle the ENABLE_IRONIC flag without touching the environment.""" + + def _set(value): + mocker.patch( + "osism.tasks.conductor.config.Config.enable_ironic", new=value, create=True + ) + + return _set + + +# --------------------------------------------------------------------------- +# Empty configuration / ironic disabled +# --------------------------------------------------------------------------- + + +def test_empty_file_returns_empty_dict_and_warns( + mocker, patch_openstack, enable_ironic, loguru_logs +): + enable_ironic("True") + _patch_open(mocker, "") + + assert get_configuration() == {} + patch_openstack.image_get.assert_not_called() + patch_openstack.network_get.assert_not_called() + assert _has_log(loguru_logs, "WARNING", "conductor configuration is empty") + + +def test_ironic_disabled_returns_yaml_untouched(mocker, patch_openstack, enable_ironic): + enable_ironic("False") + payload = { + "ironic_parameters": { + "instance_info": {"image_source": "ubuntu"}, + "driver_info": {"deploy_kernel": "kernel"}, + } + } + _patch_open(mocker, payload) + + assert get_configuration() == payload + patch_openstack.image_get.assert_not_called() + patch_openstack.network_get.assert_not_called() + + +@pytest.mark.parametrize("flag", ["false", "no", "0", "off", "anything-else", ""]) +def test_ironic_disabled_for_various_falsy_strings( + mocker, patch_openstack, enable_ironic, flag +): + enable_ironic(flag) + _patch_open(mocker, {"foo": "bar"}) + + assert get_configuration() == {"foo": "bar"} + patch_openstack.image_get.assert_not_called() + + +@pytest.mark.parametrize("flag", ["true", "True", "TRUE", "yes", "YES", "Yes"]) +def test_ironic_enabled_for_truthy_strings_processes_config( + mocker, patch_openstack, enable_ironic, flag +): + enable_ironic(flag) + payload = { + "ironic_parameters": { + "instance_info": {"image_source": "ubuntu"}, + } + } + _patch_open(mocker, payload) + patch_openstack.image_get.return_value = SimpleNamespace(id="image-uuid") + + result = get_configuration() + + patch_openstack.image_get.assert_called_once_with("ubuntu") + assert result["ironic_parameters"]["instance_info"]["image_source"] == "image-uuid" + + +def test_ironic_enabled_without_ironic_parameters_returns_as_is_and_logs_error( + mocker, patch_openstack, enable_ironic, loguru_logs +): + enable_ironic("True") + payload = {"unrelated": {"key": "value"}} + _patch_open(mocker, payload) + + assert get_configuration() == payload + patch_openstack.image_get.assert_not_called() + patch_openstack.network_get.assert_not_called() + assert _has_log( + loguru_logs, + "ERROR", + "ironic_parameters not found in the conductor configuration", + ) + + +# --------------------------------------------------------------------------- +# image_source / deploy_kernel / deploy_ramdisk resolution +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("section,key", IMAGE_FIELDS) +def test_image_field_name_resolved_to_uuid( + mocker, patch_openstack, enable_ironic, section, key +): + enable_ironic("True") + _patch_open(mocker, _payload_with(section, key, "named-image")) + patch_openstack.image_get.return_value = SimpleNamespace(id="resolved-uuid") + + result = get_configuration() + + patch_openstack.image_get.assert_called_once_with("named-image") + assert result["ironic_parameters"][section][key] == "resolved-uuid" + + +@pytest.mark.parametrize("section,key", IMAGE_FIELDS) +@pytest.mark.parametrize( + "value", + [ + pytest.param("11111111-1111-1111-1111-111111111111", id="uuid"), + pytest.param("http://example.com/images/file.qcow2", id="url"), + ], +) +def test_image_field_uuid_or_url_pass_through( + mocker, patch_openstack, enable_ironic, section, key, value +): + enable_ironic("True") + _patch_open(mocker, _payload_with(section, key, value)) + + result = get_configuration() + + patch_openstack.image_get.assert_not_called() + assert result["ironic_parameters"][section][key] == value + + +@pytest.mark.parametrize("section,key", IMAGE_FIELDS) +def test_image_field_unresolved_keeps_original_and_warns( + mocker, patch_openstack, enable_ironic, loguru_logs, section, key +): + enable_ironic("True") + _patch_open(mocker, _payload_with(section, key, "missing-image")) + patch_openstack.image_get.return_value = None + + result = get_configuration() + + assert result["ironic_parameters"][section][key] == "missing-image" + assert _has_log( + loguru_logs, + "WARNING", + "Could not resolve image ID for missing-image", + ) + + +def test_instance_info_present_without_image_source_is_noop( + mocker, patch_openstack, enable_ironic +): + enable_ironic("True") + _patch_open( + mocker, + {"ironic_parameters": {"instance_info": {"other_key": "value"}}}, + ) + + result = get_configuration() + + patch_openstack.image_get.assert_not_called() + assert result["ironic_parameters"]["instance_info"] == {"other_key": "value"} + + +# --------------------------------------------------------------------------- +# cleaning_network / provisioning_network resolution +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("key", NETWORK_FIELDS) +def test_network_field_name_resolved_to_id(mocker, patch_openstack, enable_ironic, key): + enable_ironic("True") + _patch_open(mocker, _payload_with("driver_info", key, "named-network")) + patch_openstack.network_get.return_value = SimpleNamespace(id="resolved-net-id") + + result = get_configuration() + + patch_openstack.network_get.assert_called_once_with("named-network") + assert result["ironic_parameters"]["driver_info"][key] == "resolved-net-id" + + +@pytest.mark.parametrize("key", NETWORK_FIELDS) +def test_network_field_unresolved_keeps_original_and_warns( + mocker, patch_openstack, enable_ironic, loguru_logs, key +): + enable_ironic("True") + _patch_open(mocker, _payload_with("driver_info", key, "missing-network")) + patch_openstack.network_get.return_value = None + + result = get_configuration() + + assert result["ironic_parameters"]["driver_info"][key] == "missing-network" + assert _has_log( + loguru_logs, + "WARNING", + "Could not resolve network ID for missing-network", + ) + + +# --------------------------------------------------------------------------- +# Empty/missing sections +# --------------------------------------------------------------------------- + + +def test_empty_ironic_parameters_processed_without_calls( + mocker, patch_openstack, enable_ironic +): + enable_ironic("True") + _patch_open(mocker, {"ironic_parameters": {}}) + + result = get_configuration() + + assert result == {"ironic_parameters": {}} + patch_openstack.image_get.assert_not_called() + patch_openstack.network_get.assert_not_called() + + +def test_driver_info_present_without_supported_keys_is_noop( + mocker, patch_openstack, enable_ironic +): + enable_ironic("True") + payload = {"ironic_parameters": {"driver_info": {"unrelated": "value"}}} + _patch_open(mocker, payload) + + result = get_configuration() + + assert result == payload + patch_openstack.image_get.assert_not_called() + patch_openstack.network_get.assert_not_called() + + +def test_full_configuration_resolves_all_fields(mocker, patch_openstack, enable_ironic): + enable_ironic("True") + payload = { + "ironic_parameters": { + "instance_info": {"image_source": "ubuntu"}, + "driver_info": { + "deploy_kernel": "kernel", + "deploy_ramdisk": "ramdisk", + "cleaning_network": "cleaning", + "provisioning_network": "provisioning", + }, + } + } + _patch_open(mocker, payload) + patch_openstack.image_get.side_effect = [ + SimpleNamespace(id="image-uuid"), + SimpleNamespace(id="kernel-uuid"), + SimpleNamespace(id="ramdisk-uuid"), + ] + patch_openstack.network_get.side_effect = [ + SimpleNamespace(id="cleaning-id"), + SimpleNamespace(id="provisioning-id"), + ] + + result = get_configuration() + + assert result == { + "ironic_parameters": { + "instance_info": {"image_source": "image-uuid"}, + "driver_info": { + "deploy_kernel": "kernel-uuid", + "deploy_ramdisk": "ramdisk-uuid", + "cleaning_network": "cleaning-id", + "provisioning_network": "provisioning-id", + }, + } + } + assert patch_openstack.image_get.call_count == 3 + assert patch_openstack.network_get.call_count == 2 + + +# --------------------------------------------------------------------------- +# File location +# --------------------------------------------------------------------------- + + +def test_reads_configuration_from_etc_conductor_yml( + mocker, patch_openstack, enable_ironic +): + enable_ironic("False") + opener = _patch_open(mocker, {"foo": "bar"}) + + get_configuration() + + opener.assert_called_once_with("/etc/conductor.yml") + + +# --------------------------------------------------------------------------- +# Module sanity +# --------------------------------------------------------------------------- + + +def test_module_exposes_get_configuration(): + assert callable(config_module.get_configuration) diff --git a/tests/unit/tasks/conductor/test_netbox.py b/tests/unit/tasks/conductor/test_netbox.py new file mode 100644 index 000000000..dc48462aa --- /dev/null +++ b/tests/unit/tasks/conductor/test_netbox.py @@ -0,0 +1,444 @@ +# SPDX-License-Identifier: Apache-2.0 + +from types import SimpleNamespace + +import pytest +import yaml + +from osism.tasks.conductor.netbox import ( + get_device_oob_ip, + get_nb_device_query_list_ironic, + get_nb_device_query_list_sonic, +) + +# --------------------------------------------------------------------------- +# Helpers / fixtures +# --------------------------------------------------------------------------- + + +# The two query-list helpers are structurally identical; parametrize over them +# instead of duplicating every test. Each variant carries the setting attribute +# it reads and the function under test. +QUERY_LIST_VARIANTS = [ + pytest.param( + "NETBOX_FILTER_CONDUCTOR_IRONIC", + get_nb_device_query_list_ironic, + id="ironic", + ), + pytest.param( + "NETBOX_FILTER_CONDUCTOR_SONIC", + get_nb_device_query_list_sonic, + id="sonic", + ), +] + + +def _has_log(records, level, substring): + return any(r["level"] == level and substring in r["message"] for r in records) + + +@pytest.fixture +def patch_filter_setting(mocker): + """Patch one of the ``NETBOX_FILTER_CONDUCTOR_*`` settings.""" + + def _set(setting_name, value): + mocker.patch(f"osism.tasks.conductor.netbox.settings.{setting_name}", new=value) + + return _set + + +@pytest.fixture +def patch_location_lookups(mocker): + location = mocker.patch("osism.tasks.conductor.netbox.netbox.get_location_id") + rack = mocker.patch("osism.tasks.conductor.netbox.netbox.get_rack_id") + return SimpleNamespace(get_location_id=location, get_rack_id=rack) + + +@pytest.fixture +def mock_nb(mocker): + """Replace ``osism.utils.nb`` (lazy attribute) with a fresh MagicMock.""" + nb = mocker.MagicMock() + mocker.patch("osism.utils.nb", new=nb, create=True) + return nb + + +class _IPRecord: + """Stand-in for a pynetbox IP record. + + ``ipaddress.ip_interface`` accepts strings, so the function under test + relies on pynetbox records being stringified to the address. The simple + ``SimpleNamespace`` shim used elsewhere does not satisfy that contract. + """ + + def __init__(self, address): + self.address = address + + def __str__(self): + return self.address + + +def _make_device( + name="dev", + device_id=1, + oob_ip=None, +): + return SimpleNamespace(name=name, id=device_id, oob_ip=oob_ip) + + +def _make_interface(interface_id, mgmt_only=False): + return SimpleNamespace(id=interface_id, mgmt_only=mgmt_only) + + +def _make_ip(address): + return _IPRecord(address) + + +# --------------------------------------------------------------------------- +# get_nb_device_query_list_{ironic,sonic} – happy paths +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_simple_filters_pass_through(patch_filter_setting, setting, query_func): + patch_filter_setting(setting, "- site: dc1\n- tag: ironic\n") + + assert query_func() == [{"site": "dc1"}, {"tag": "ironic"}] + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_all_supported_pass_through_filters_unchanged( + patch_filter_setting, patch_location_lookups, setting, query_func +): + patch_filter_setting( + setting, + "- site: dc1\n" + "- region: eu-west\n" + "- site_group: prod\n" + "- tag: ironic\n" + "- status: active\n", + ) + + assert query_func() == [ + {"site": "dc1"}, + {"region": "eu-west"}, + {"site_group": "prod"}, + {"tag": "ironic"}, + {"status": "active"}, + ] + patch_location_lookups.get_location_id.assert_not_called() + patch_location_lookups.get_rack_id.assert_not_called() + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_location_resolved_and_renamed( + patch_filter_setting, patch_location_lookups, setting, query_func +): + patch_filter_setting(setting, "- location: dc1-room-3\n") + patch_location_lookups.get_location_id.return_value = 17 + + assert query_func() == [{"location_id": 17}] + patch_location_lookups.get_location_id.assert_called_once_with("dc1-room-3") + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_rack_resolved_and_renamed( + patch_filter_setting, patch_location_lookups, setting, query_func +): + patch_filter_setting(setting, "- rack: r42\n") + patch_location_lookups.get_rack_id.return_value = 99 + + assert query_func() == [{"rack_id": 99}] + patch_location_lookups.get_rack_id.assert_called_once_with("r42") + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_location_and_rack_combined( + patch_filter_setting, patch_location_lookups, setting, query_func +): + patch_filter_setting(setting, "- location: room-1\n rack: r1\n") + patch_location_lookups.get_location_id.return_value = 5 + patch_location_lookups.get_rack_id.return_value = 11 + + assert query_func() == [{"location_id": 5, "rack_id": 11}] + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_empty_list_returns_empty(patch_filter_setting, setting, query_func): + patch_filter_setting(setting, "[]\n") + + assert query_func() == [] + + +# --------------------------------------------------------------------------- +# get_nb_device_query_list_{ironic,sonic} – error paths +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_invalid_yaml_raises_yamlerror_and_logs( + patch_filter_setting, loguru_logs, setting, query_func +): + patch_filter_setting(setting, "foo: [unterminated\n") + + with pytest.raises(yaml.YAMLError): + query_func() + assert _has_log( + loguru_logs, + "ERROR", + f"Setting {setting} needs to be an array of mappings", + ) + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_non_list_raises_typeerror_and_logs( + patch_filter_setting, loguru_logs, setting, query_func +): + patch_filter_setting(setting, "foo: bar\n") + + with pytest.raises(TypeError): + query_func() + assert _has_log( + loguru_logs, + "ERROR", + f"Setting {setting} needs to be an array of mappings", + ) + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_string_element_raises_typeerror_and_logs( + patch_filter_setting, loguru_logs, setting, query_func +): + patch_filter_setting(setting, "- foo\n") + + with pytest.raises(TypeError): + query_func() + assert _has_log( + loguru_logs, + "ERROR", + f"Setting {setting} needs to be an array of mappings", + ) + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_unsupported_filter_raises_valueerror_and_logs( + patch_filter_setting, loguru_logs, setting, query_func +): + patch_filter_setting(setting, "- manufacturer: dell\n") + + with pytest.raises(ValueError): + query_func() + assert _has_log(loguru_logs, "ERROR", f"Unknown value in {setting}") + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_location_unresolved_raises_valueerror_and_logs( + patch_filter_setting, patch_location_lookups, loguru_logs, setting, query_func +): + patch_filter_setting(setting, "- location: missing\n") + patch_location_lookups.get_location_id.return_value = None + + with pytest.raises(ValueError): + query_func() + assert _has_log(loguru_logs, "ERROR", f"Unknown value in {setting}") + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_rack_unresolved_raises_valueerror_and_logs( + patch_filter_setting, patch_location_lookups, loguru_logs, setting, query_func +): + patch_filter_setting(setting, "- rack: missing\n") + patch_location_lookups.get_rack_id.return_value = None + + with pytest.raises(ValueError): + query_func() + assert _has_log(loguru_logs, "ERROR", f"Unknown value in {setting}") + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_null_setting_raises_typeerror(patch_filter_setting, setting, query_func): + """A YAML payload of ``null`` parses to ``None`` (not a list).""" + patch_filter_setting(setting, "") + + with pytest.raises(TypeError): + query_func() + + +@pytest.mark.parametrize("setting,query_func", QUERY_LIST_VARIANTS) +def test_list_with_null_element_raises_typeerror( + patch_filter_setting, setting, query_func +): + """A list element that is ``None`` is not a dict.""" + patch_filter_setting(setting, "- null\n") + + with pytest.raises(TypeError): + query_func() + + +# --------------------------------------------------------------------------- +# get_device_oob_ip – oob_ip set on device +# --------------------------------------------------------------------------- + + +def test_oob_ip_uses_oob_field_when_set(mock_nb): + device = _make_device(oob_ip=_IPRecord("10.0.0.5/24")) + + assert get_device_oob_ip(device) == ("10.0.0.5", 24) + mock_nb.dcim.interfaces.filter.assert_not_called() + mock_nb.ipam.ip_addresses.filter.assert_not_called() + + +def test_oob_ip_falsy_oob_ip_falls_back_to_mgmt(mock_nb): + """``oob_ip = None`` → fall back to mgmt-only interface lookup.""" + device = _make_device(device_id=7, oob_ip=None) + iface = _make_interface(interface_id=42, mgmt_only=True) + mock_nb.dcim.interfaces.filter.return_value = [iface] + mock_nb.ipam.ip_addresses.filter.return_value = [_make_ip("10.0.0.6/24")] + + assert get_device_oob_ip(device) == ("10.0.0.6", 24) + mock_nb.dcim.interfaces.filter.assert_called_once_with(device_id=7) + mock_nb.ipam.ip_addresses.filter.assert_called_once_with(assigned_object_id=42) + + +def test_oob_ip_skips_non_mgmt_interfaces(mock_nb): + device = _make_device(device_id=8) + non_mgmt = _make_interface(interface_id=1, mgmt_only=False) + mgmt = _make_interface(interface_id=2, mgmt_only=True) + mock_nb.dcim.interfaces.filter.return_value = [non_mgmt, mgmt] + mock_nb.ipam.ip_addresses.filter.return_value = [_make_ip("192.168.0.10/16")] + + assert get_device_oob_ip(device) == ("192.168.0.10", 16) + # Only the mgmt-only interface should have triggered an IP lookup. + mock_nb.ipam.ip_addresses.filter.assert_called_once_with(assigned_object_id=2) + + +def test_oob_ip_first_mgmt_without_ips_falls_through_to_second(mock_nb): + device = _make_device(device_id=9) + first_mgmt = _make_interface(interface_id=11, mgmt_only=True) + second_mgmt = _make_interface(interface_id=22, mgmt_only=True) + mock_nb.dcim.interfaces.filter.return_value = [first_mgmt, second_mgmt] + mock_nb.ipam.ip_addresses.filter.side_effect = [ + [], + [_make_ip("10.0.0.7/24")], + ] + + assert get_device_oob_ip(device) == ("10.0.0.7", 24) + assert mock_nb.ipam.ip_addresses.filter.call_count == 2 + + +def test_oob_ip_no_mgmt_interfaces_returns_none(mock_nb): + device = _make_device(device_id=1) + mock_nb.dcim.interfaces.filter.return_value = [ + _make_interface(interface_id=1, mgmt_only=False), + _make_interface(interface_id=2, mgmt_only=False), + ] + + assert get_device_oob_ip(device) is None + mock_nb.ipam.ip_addresses.filter.assert_not_called() + + +def test_oob_ip_empty_interfaces_returns_none(mock_nb): + device = _make_device(device_id=1) + mock_nb.dcim.interfaces.filter.return_value = [] + + assert get_device_oob_ip(device) is None + mock_nb.ipam.ip_addresses.filter.assert_not_called() + + +def test_oob_ip_mgmt_interface_without_ips_returns_none(mock_nb): + device = _make_device(device_id=1) + iface = _make_interface(interface_id=42, mgmt_only=True) + mock_nb.dcim.interfaces.filter.return_value = [iface] + mock_nb.ipam.ip_addresses.filter.return_value = [] + + assert get_device_oob_ip(device) is None + + +def test_oob_ip_ip_with_blank_address_skipped(mock_nb): + device = _make_device(device_id=1) + iface = _make_interface(interface_id=42, mgmt_only=True) + mock_nb.dcim.interfaces.filter.return_value = [iface] + mock_nb.ipam.ip_addresses.filter.return_value = [_make_ip("")] + + assert get_device_oob_ip(device) is None + + +def test_oob_ip_ipv6_returns_address_and_prefix(mock_nb): + device = _make_device(oob_ip=_IPRecord("2001:db8::1/64")) + + assert get_device_oob_ip(device) == ("2001:db8::1", 64) + + +def test_oob_ip_malformed_oob_returns_none_and_warns(mock_nb, loguru_logs): + device = _make_device(name="dev-malformed", oob_ip=_IPRecord("not-an-ip")) + + assert get_device_oob_ip(device) is None + assert _has_log( + loguru_logs, + "WARNING", + "Could not get OOB IP for device dev-malformed", + ) + + +def test_oob_ip_interfaces_filter_raises_returns_none_and_warns(mock_nb, loguru_logs): + device = _make_device(name="dev-down", device_id=1, oob_ip=None) + mock_nb.dcim.interfaces.filter.side_effect = RuntimeError("netbox down") + + assert get_device_oob_ip(device) is None + assert _has_log( + loguru_logs, + "WARNING", + "Could not get OOB IP for device dev-down", + ) + + +def test_oob_ip_ip_addresses_filter_raises_returns_none_and_warns(mock_nb, loguru_logs): + device = _make_device(name="dev-ipam", device_id=1, oob_ip=None) + iface = _make_interface(interface_id=42, mgmt_only=True) + mock_nb.dcim.interfaces.filter.return_value = [iface] + mock_nb.ipam.ip_addresses.filter.side_effect = RuntimeError("ipam down") + + assert get_device_oob_ip(device) is None + assert _has_log( + loguru_logs, + "WARNING", + "Could not get OOB IP for device dev-ipam", + ) + + +def test_oob_ip_uses_keyword_arguments_for_netbox_filters(mock_nb): + device = _make_device(device_id=123, oob_ip=None) + iface = _make_interface(interface_id=456, mgmt_only=True) + mock_nb.dcim.interfaces.filter.return_value = [iface] + mock_nb.ipam.ip_addresses.filter.return_value = [_make_ip("10.0.0.8/24")] + + get_device_oob_ip(device) + + interfaces_call = mock_nb.dcim.interfaces.filter.call_args + assert interfaces_call.args == () + assert interfaces_call.kwargs == {"device_id": 123} + + ip_call = mock_nb.ipam.ip_addresses.filter.call_args + assert ip_call.args == () + assert ip_call.kwargs == {"assigned_object_id": 456} + + +def test_oob_ip_device_without_oob_ip_attribute_falls_back(mock_nb): + """``hasattr(device, 'oob_ip')`` is False → mgmt-interface path runs.""" + device = SimpleNamespace(name="dev-no-oob-attr", id=5) + iface = _make_interface(interface_id=1, mgmt_only=True) + mock_nb.dcim.interfaces.filter.return_value = [iface] + mock_nb.ipam.ip_addresses.filter.return_value = [_make_ip("10.0.0.9/24")] + + assert get_device_oob_ip(device) == ("10.0.0.9", 24) + + +def test_oob_ip_first_assigned_address_wins(mock_nb): + device = _make_device(device_id=1, oob_ip=None) + iface = _make_interface(interface_id=42, mgmt_only=True) + mock_nb.dcim.interfaces.filter.return_value = [iface] + mock_nb.ipam.ip_addresses.filter.return_value = [ + _make_ip("10.0.0.10/24"), + _make_ip("10.0.0.11/24"), + ] + + assert get_device_oob_ip(device) == ("10.0.0.10", 24)