From 639ce706890e86dd0bd3976a975d0f87db775db9 Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Tue, 28 Apr 2026 18:27:59 +0200 Subject: [PATCH] Add unit tests for SONiC config_generator orchestrator and service caches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Covers natural_sort_key, generate_sonic_config (base-config loading, deepcopy, primary-IP / AS routing, OOB management, breakout merge, config_version normalization), the metalbox / DNS caches, and the log-server / SNMP helpers. Helpers from companion sub-issues are patched at their import sites so this file exercises only the orchestrator's own glue. Also hardens generate_sonic_config so the top-level scaffold keys it and its helpers index into directly are always present, even when /etc/sonic/config_db.json is absent or fails to load. The scaffold key list is exposed as TOP_LEVEL_SCAFFOLD_KEYS and reused by the test helper so the production code and the tests cannot drift apart when keys are added or removed. Extends the tuple with BGP_GLOBALS_AF, BGP_GLOBALS_AF_NETWORK, BGP_GLOBALS_ROUTE_ADVERTISE, BGP_NEIGHBOR, BGP_NEIGHBOR_AF, VXLAN_TUNNEL, VXLAN_EVPN_NVO and VXLAN_TUNNEL_MAP, all of which the orchestrator writes into directly and which previously raised KeyError on devices with connected ports, Loopback0 routes or VRFs with VNI when config_db.json was missing or sparse. Removes the orphaned NetBox-crawl NTP path (_get_ntp_servers, _ntp_servers_cache, clear_ntp_cache); _add_ntp_configuration reaches the metalbox IP via _get_metalbox_ip_for_device directly, so the crawl helper and its cache had no production callers. The test module is split per concern — orchestrator, metalbox cache, NTP, DNS, log-server, SNMP — into sibling test_config_generator_*.py files sharing a small _config_generator_helpers.py module, with the cache-reset and mock_nb fixtures lifted into conftest.py and opted into via pytestmark.usefixtures so unrelated SONiC tests stay unaffected. Closes #2221 AI-assisted: Claude Code Signed-off-by: Christian Berendt --- .../tasks/conductor/sonic/config_generator.py | 109 ++-- .../sonic/_config_generator_helpers.py | 81 +++ tests/unit/tasks/conductor/sonic/conftest.py | 31 + .../sonic/test_config_generator_dns.py | 49 ++ .../sonic/test_config_generator_log.py | 69 +++ .../test_config_generator_metalbox_cache.py | 210 +++++++ .../sonic/test_config_generator_ntp.py | 58 ++ .../test_config_generator_orchestrator.py | 556 ++++++++++++++++++ .../sonic/test_config_generator_snmp.py | 171 ++++++ 9 files changed, 1265 insertions(+), 69 deletions(-) create mode 100644 tests/unit/tasks/conductor/sonic/_config_generator_helpers.py create mode 100644 tests/unit/tasks/conductor/sonic/test_config_generator_dns.py create mode 100644 tests/unit/tasks/conductor/sonic/test_config_generator_log.py create mode 100644 tests/unit/tasks/conductor/sonic/test_config_generator_metalbox_cache.py create mode 100644 tests/unit/tasks/conductor/sonic/test_config_generator_ntp.py create mode 100644 tests/unit/tasks/conductor/sonic/test_config_generator_orchestrator.py create mode 100644 tests/unit/tasks/conductor/sonic/test_config_generator_snmp.py diff --git a/osism/tasks/conductor/sonic/config_generator.py b/osism/tasks/conductor/sonic/config_generator.py index 31fd43db..1e1575d8 100644 --- a/osism/tasks/conductor/sonic/config_generator.py +++ b/osism/tasks/conductor/sonic/config_generator.py @@ -40,9 +40,6 @@ from .cache import get_cached_device_interfaces from .constants import BGP_AF_L2VPN_EVPN_TAG, DEFAULT_SONIC_ROLES -# Global cache for NTP servers to avoid multiple queries -_ntp_servers_cache = None - # Global cache for metalbox IPs per device to avoid duplicate lookups _metalbox_ip_cache: dict[int, Optional[str]] = {} @@ -52,6 +49,40 @@ # VXLAN VTEP name used for VXLAN tunnel configuration VXLAN_VTEP_NAME = "vtepServ" +# Top-level scaffold keys that the orchestrator and downstream helpers index +# into directly. Kept as a single source of truth so that the production code +# and the test helpers cannot drift apart when keys are added or removed. +TOP_LEVEL_SCAFFOLD_KEYS = ( + "DEVICE_METADATA", + "BGP_GLOBALS", + "BGP_GLOBALS_AF", + "BGP_GLOBALS_AF_NETWORK", + "BGP_GLOBALS_ROUTE_ADVERTISE", + "BGP_NEIGHBOR", + "BGP_NEIGHBOR_AF", + "MGMT_INTERFACE", + "STATIC_ROUTE", + "BREAKOUT_CFG", + "BREAKOUT_PORTS", + "NTP_SERVER", + "DNS_NAMESERVER", + "PORT", + "INTERFACE", + "VLAN", + "VLAN_INTERFACE", + "VLAN_MEMBER", + "LOOPBACK", + "LOOPBACK_INTERFACE", + "PORTCHANNEL", + "PORTCHANNEL_INTERFACE", + "PORTCHANNEL_MEMBER", + "VRF", + "VXLAN_TUNNEL", + "VXLAN_EVPN_NVO", + "VXLAN_TUNNEL_MAP", + "VERSIONS", +) + def natural_sort_key(port_name): """Extract numeric part from port name for natural sorting.""" @@ -160,6 +191,12 @@ def generate_sonic_config(device, hwsku, device_as_mapping=None, config_version= # Ensure we start fresh even on error config = {} + # Ensure the top-level scaffold keys the orchestrator and downstream + # helpers index into directly are always present, even when the on-disk + # base config is missing or only partially populated. + for _scaffold_key in TOP_LEVEL_SCAFFOLD_KEYS: + config.setdefault(_scaffold_key, {}) + # Update DEVICE_METADATA with NetBox information if "localhost" not in config["DEVICE_METADATA"]: config["DEVICE_METADATA"]["localhost"] = {} @@ -1563,64 +1600,6 @@ def _get_metalbox_ip_for_device(device): return None -def _get_ntp_servers(): - """Get NTP servers from manager/metalbox devices. Uses caching to avoid repeated queries.""" - global _ntp_servers_cache - - if _ntp_servers_cache is not None: - logger.debug("Using cached NTP servers") - return _ntp_servers_cache - - ntp_servers = {} - try: - # Get devices with manager or metalbox device roles - devices_manager = utils.nb.dcim.devices.filter(role="manager") - devices_metalbox = utils.nb.dcim.devices.filter(role="metalbox") - - # Combine both device lists - ntp_devices = list(devices_manager) + list(devices_metalbox) - logger.debug(f"Found {len(ntp_devices)} potential NTP devices") - - for ntp_device in ntp_devices: - # Get interfaces for this device to find Loopback0 - device_interfaces = utils.nb.dcim.interfaces.filter(device_id=ntp_device.id) - - for interface in device_interfaces: - # Look for Loopback0 interface - if interface.name == "Loopback0": - # Get IP addresses assigned to this Loopback0 interface - ip_addresses = utils.nb.ipam.ip_addresses.filter( - assigned_object_id=interface.id, - ) - - for ip_addr in ip_addresses: - if ip_addr.address: - # Extract just the IPv4 address without prefix - ip_only = ip_addr.address.split("/")[0] - - # Check if it's an IPv4 address (simple check) - if "." in ip_only and ":" not in ip_only: - ntp_servers[ip_only] = { - "maxpoll": "10", - "minpoll": "6", - "prefer": "false", - } - logger.info( - f"Found NTP server {ip_only} from device {ntp_device.name} with role {ntp_device.role.slug}" - ) - break - - # Cache the results - _ntp_servers_cache = ntp_servers - logger.debug(f"Cached {len(ntp_servers)} NTP servers") - - except Exception as e: - logger.warning(f"Could not process NTP servers: {e}") - _ntp_servers_cache = {} - - return _ntp_servers_cache - - def _add_ntp_configuration(config, device): """Add NTP_SERVER configuration to device config. @@ -1646,13 +1625,6 @@ def _add_ntp_configuration(config, device): logger.warning(f"Could not add NTP configuration to device {device.name}: {e}") -def clear_ntp_cache(): - """Clear the NTP servers cache. Should be called at the start of sync_sonic.""" - global _ntp_servers_cache - _ntp_servers_cache = None - logger.debug("Cleared NTP servers cache") - - def clear_metalbox_ip_cache(): """Clear the metalbox IP cache. Should be called at the start of sync_sonic.""" global _metalbox_ip_cache @@ -1690,7 +1662,6 @@ def _add_dns_configuration(config, device): def clear_all_caches(): """Clear all caches in config_generator module.""" - clear_ntp_cache() clear_metalbox_ip_cache() clear_metalbox_devices_cache() clear_port_config_cache() diff --git a/tests/unit/tasks/conductor/sonic/_config_generator_helpers.py b/tests/unit/tasks/conductor/sonic/_config_generator_helpers.py new file mode 100644 index 00000000..b7139882 --- /dev/null +++ b/tests/unit/tasks/conductor/sonic/_config_generator_helpers.py @@ -0,0 +1,81 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Shared helpers for the SONiC ``config_generator`` unit-test split. + +The leading underscore keeps pytest from collecting this module as a test file. +""" + +import json +from types import SimpleNamespace +from unittest.mock import mock_open + +from osism.tasks.conductor.sonic import config_generator +from osism.tasks.conductor.sonic.config_generator import TOP_LEVEL_SCAFFOLD_KEYS + + +def make_base_config(version=None): + """Build a base ``config_db.json`` scaffold with every top-level key the + orchestrator (and its mocked helpers) index into directly. + + The key list is sourced from the production-side ``TOP_LEVEL_SCAFFOLD_KEYS`` + constant so the helper cannot drift when keys are added or removed. + Optionally seeds ``VERSIONS.DATABASE.VERSION`` so the version-handling + branch can be exercised explicitly. + """ + + cfg = {key: {} for key in TOP_LEVEL_SCAFFOLD_KEYS} + if version is not None: + cfg["VERSIONS"] = {"DATABASE": {"VERSION": version}} + return cfg + + +def patch_base_config(mocker, *, exists=True, base_config=None, raise_on_open=None): + """Patch ``os.path.exists`` and ``builtins.open`` for the base-config load. + + - ``exists=True`` and ``base_config`` provided → ``open`` returns that + JSON-encoded scaffold. + - ``exists=False`` → ``open`` is not patched (the orchestrator never + reaches the ``with open`` path). + - ``raise_on_open`` → ``open`` raises this exception (e.g. ``OSError``). + """ + + mocker.patch.object(config_generator.os.path, "exists", return_value=exists) + if not exists: + return + if raise_on_open is not None: + mocker.patch("builtins.open", side_effect=raise_on_open) + return + cfg = base_config if base_config is not None else make_base_config() + mocker.patch("builtins.open", mock_open(read_data=json.dumps(cfg))) + + +def make_iface(name, *, mgmt_only=False, type_value=None, iface_id=None): + return SimpleNamespace( + id=iface_id if iface_id is not None else id(object()), + name=name, + mgmt_only=mgmt_only, + type=SimpleNamespace(value=type_value) if type_value is not None else None, + ) + + +def make_ip(address): + return SimpleNamespace(address=address) + + +def seed_metalbox_cache(metalbox_id=10, name="mb", *, interfaces): + """Set ``_metalbox_devices_cache`` directly (skip ``_load`` to keep tests + fast and independent of NetBox-shape concerns). + + ``interfaces`` is a list of ``(interface_obj, is_vlan, [ip_strings])``. + """ + cache_entry = { + "device": SimpleNamespace(id=metalbox_id, name=name), + "interfaces": {}, + } + for iface, is_vlan, addresses in interfaces: + cache_entry["interfaces"][iface.id] = { + "interface": iface, + "is_vlan": is_vlan, + "ips": [make_ip(addr) for addr in addresses], + } + config_generator._metalbox_devices_cache = {metalbox_id: cache_entry} diff --git a/tests/unit/tasks/conductor/sonic/conftest.py b/tests/unit/tasks/conductor/sonic/conftest.py index b99c9818..4596dec2 100644 --- a/tests/unit/tasks/conductor/sonic/conftest.py +++ b/tests/unit/tasks/conductor/sonic/conftest.py @@ -3,10 +3,41 @@ """Shared fixtures for the SONiC unit tests.""" from types import SimpleNamespace +from unittest.mock import MagicMock import pytest +@pytest.fixture +def reset_config_generator_caches(): + """Reset every module global the ``config_generator`` orchestrator touches. + + Without this, a previous test's ``_metalbox_ip_cache`` / + ``_metalbox_devices_cache`` would leak into the next one and make the + suite order-dependent. Files that exercise ``config_generator`` opt in + via ``pytestmark = pytest.mark.usefixtures(...)`` so the reset never runs + for unrelated SONiC tests. + """ + from osism.tasks.conductor.sonic import config_generator + + config_generator.clear_all_caches() + yield + config_generator.clear_all_caches() + + +@pytest.fixture +def mock_nb(mocker): + """Replace ``utils.nb`` for the duration of one test. + + ``utils.nb`` is normally lazily wired through ``__getattr__`` and would + try to reach a real NetBox instance — ``create=True`` is needed because + the attribute may not be bound yet. + """ + nb = MagicMock() + mocker.patch("osism.utils.nb", new=nb, create=True) + return nb + + @pytest.fixture def make_interface(): """Build a minimal NetBox-shaped interface stub.""" diff --git a/tests/unit/tasks/conductor/sonic/test_config_generator_dns.py b/tests/unit/tasks/conductor/sonic/test_config_generator_dns.py new file mode 100644 index 00000000..2af6c2de --- /dev/null +++ b/tests/unit/tasks/conductor/sonic/test_config_generator_dns.py @@ -0,0 +1,49 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for ``_add_dns_configuration`` in ``config_generator``.""" + +from types import SimpleNamespace + +import pytest + +from osism.tasks.conductor.sonic import config_generator +from osism.tasks.conductor.sonic.config_generator import _add_dns_configuration + +pytestmark = pytest.mark.usefixtures("reset_config_generator_caches") + + +def test_add_dns_configuration_with_metalbox_ip(mocker): + mocker.patch.object( + config_generator, + "_get_metalbox_ip_for_device", + return_value="10.0.0.1", + ) + config = {"DNS_NAMESERVER": {}} + + _add_dns_configuration(config, SimpleNamespace(name="leaf-1")) + + assert config["DNS_NAMESERVER"] == {"10.0.0.1": {}} + + +def test_add_dns_configuration_no_metalbox_ip(mocker): + mocker.patch.object( + config_generator, "_get_metalbox_ip_for_device", return_value=None + ) + config = {"DNS_NAMESERVER": {}} + + _add_dns_configuration(config, SimpleNamespace(name="leaf-1")) + + assert config["DNS_NAMESERVER"] == {} + + +def test_add_dns_configuration_helper_exception_swallowed(mocker): + mocker.patch.object( + config_generator, + "_get_metalbox_ip_for_device", + side_effect=RuntimeError("kaboom"), + ) + config = {"DNS_NAMESERVER": {}} + + _add_dns_configuration(config, SimpleNamespace(name="leaf-1")) + + assert config["DNS_NAMESERVER"] == {} diff --git a/tests/unit/tasks/conductor/sonic/test_config_generator_log.py b/tests/unit/tasks/conductor/sonic/test_config_generator_log.py new file mode 100644 index 00000000..f5f4df03 --- /dev/null +++ b/tests/unit/tasks/conductor/sonic/test_config_generator_log.py @@ -0,0 +1,69 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for ``_add_log_server_configuration`` in ``config_generator``.""" + +from types import SimpleNamespace + +import pytest + +from osism.tasks.conductor.sonic.config_generator import _add_log_server_configuration + +pytestmark = pytest.mark.usefixtures("reset_config_generator_caches") + + +def _device_with_ctx(**ctx): + return SimpleNamespace(name="leaf-1", config_context=ctx) + + +def test_add_log_server_configuration_defaults(): + config = {} + device = _device_with_ctx(_segment_log_server_hosts=["10.1.1.1", "10.1.1.2"]) + + _add_log_server_configuration(config, device) + + for host in ("10.1.1.1", "10.1.1.2"): + assert config["SYSLOG_SERVER"][host] == { + "message-type": "log", + "protocol": "UDP", + "remote-port": "514", + "severity": "info", + "vrf_name": "mgmt", + } + + +def test_add_log_server_configuration_protocol_uppercased(): + config = {} + device = _device_with_ctx( + _segment_log_server_hosts=["10.1.1.1"], + _segment_log_server_proto="tcp", + ) + + _add_log_server_configuration(config, device) + + assert config["SYSLOG_SERVER"]["10.1.1.1"]["protocol"] == "TCP" + + +def test_add_log_server_configuration_custom_severity_and_vrf(): + config = {} + device = _device_with_ctx( + _segment_log_server_hosts=["10.1.1.1"], + _segment_log_server_severity="debug", + _segment_log_server_vrf="default", + ) + + _add_log_server_configuration(config, device) + + entry = config["SYSLOG_SERVER"]["10.1.1.1"] + assert entry["severity"] == "debug" + assert entry["vrf_name"] == "default" + + +@pytest.mark.parametrize("hosts_value", [[], None]) +def test_add_log_server_configuration_no_hosts_skips_section(hosts_value): + config = {} + ctx = {} if hosts_value is None else {"_segment_log_server_hosts": hosts_value} + device = _device_with_ctx(**ctx) + + _add_log_server_configuration(config, device) + + assert "SYSLOG_SERVER" not in config diff --git a/tests/unit/tasks/conductor/sonic/test_config_generator_metalbox_cache.py b/tests/unit/tasks/conductor/sonic/test_config_generator_metalbox_cache.py new file mode 100644 index 00000000..fa06fa8f --- /dev/null +++ b/tests/unit/tasks/conductor/sonic/test_config_generator_metalbox_cache.py @@ -0,0 +1,210 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the metalbox-discovery service caches in +``config_generator``: ``_load_metalbox_devices_cache`` (the NetBox crawl) and +``_get_metalbox_ip_for_device`` (the per-device subnet match).""" + +from types import SimpleNamespace + +import pytest + +from osism.tasks.conductor.sonic import config_generator +from osism.tasks.conductor.sonic.config_generator import ( + _get_metalbox_ip_for_device, + _load_metalbox_devices_cache, +) + +from ._config_generator_helpers import make_iface, make_ip, seed_metalbox_cache + +pytestmark = pytest.mark.usefixtures("reset_config_generator_caches") + + +# --------------------------------------------------------------------------- +# _load_metalbox_devices_cache +# --------------------------------------------------------------------------- + + +def test_load_metalbox_devices_cache_filters_mgmt_only_and_flags_vlan(mock_nb): + metalbox = SimpleNamespace(id=10, name="metalbox-1") + eth_iface = make_iface("Ethernet0", mgmt_only=False, iface_id=100) + mgmt_iface = make_iface("eth0", mgmt_only=True, iface_id=101) + vlan_iface = make_iface( + "Vlan100", mgmt_only=False, type_value="virtual", iface_id=102 + ) + mock_nb.dcim.devices.filter.return_value = [metalbox] + mock_nb.dcim.interfaces.filter.return_value = [ + eth_iface, + mgmt_iface, + vlan_iface, + ] + mock_nb.ipam.ip_addresses.filter.side_effect = [ + [make_ip("10.0.0.1/24")], # eth_iface + [make_ip("192.168.1.1/24")], # vlan_iface + ] + + _load_metalbox_devices_cache() + + cache = config_generator._metalbox_devices_cache + assert set(cache[10]["interfaces"].keys()) == {100, 102} + # ``is_vlan`` is the result of a short-circuited and-chain, so non-VLAN + # interfaces may carry a falsy non-bool (e.g. ``None`` when ``type`` is + # absent). Compare on truthiness, not identity. + assert not cache[10]["interfaces"][100]["is_vlan"] + assert cache[10]["interfaces"][102]["is_vlan"] + assert [ip.address for ip in cache[10]["interfaces"][100]["ips"]] == ["10.0.0.1/24"] + + +def test_load_metalbox_devices_cache_per_metalbox_interface_failure_isolated( + mock_nb, +): + """One metalbox failing to enumerate interfaces must not poison the cache + for the others — the failing entry just gets an empty ``interfaces`` dict.""" + mb_a = SimpleNamespace(id=1, name="mb-a") + mb_b = SimpleNamespace(id=2, name="mb-b") + eth = make_iface("Ethernet0", iface_id=99) + mock_nb.dcim.devices.filter.return_value = [mb_a, mb_b] + mock_nb.dcim.interfaces.filter.side_effect = [ + RuntimeError("transient"), # mb-a + [eth], # mb-b + ] + mock_nb.ipam.ip_addresses.filter.return_value = [make_ip("10.0.0.1/24")] + + _load_metalbox_devices_cache() + + cache = config_generator._metalbox_devices_cache + assert cache[1]["interfaces"] == {} + assert 99 in cache[2]["interfaces"] + + +def test_load_metalbox_devices_cache_top_level_failure_resets_cache(mock_nb): + mock_nb.dcim.devices.filter.side_effect = RuntimeError("netbox down") + + _load_metalbox_devices_cache() + + assert config_generator._metalbox_devices_cache == {} + + +def test_load_metalbox_devices_cache_filters_ip_without_address(mock_nb): + """``if ip_addr.address`` filters falsy addresses; non-empty ones survive.""" + metalbox = SimpleNamespace(id=10, name="mb") + eth = make_iface("Ethernet0", iface_id=100) + mock_nb.dcim.devices.filter.return_value = [metalbox] + mock_nb.dcim.interfaces.filter.return_value = [eth] + mock_nb.ipam.ip_addresses.filter.return_value = [ + make_ip(""), # falsy — dropped + make_ip(None), # falsy — dropped + make_ip("10.0.0.1/24"), # kept + ] + + _load_metalbox_devices_cache() + + ips = config_generator._metalbox_devices_cache[10]["interfaces"][100]["ips"] + assert [ip.address for ip in ips] == ["10.0.0.1/24"] + + +# --------------------------------------------------------------------------- +# _get_metalbox_ip_for_device +# --------------------------------------------------------------------------- + + +def test_get_metalbox_ip_oob_lookup_returns_none(mocker): + mocker.patch.object(config_generator, "get_device_oob_ip", return_value=None) + device = SimpleNamespace(id=1, name="leaf-1") + + assert _get_metalbox_ip_for_device(device) is None + # ``None`` is cached so we never re-enter the lookup for this device. + assert config_generator._metalbox_ip_cache[1] is None + + +def test_get_metalbox_ip_subnet_match_returns_ip(mocker): + mocker.patch.object( + config_generator, "get_device_oob_ip", return_value=("10.0.0.5", 24) + ) + eth = make_iface("Ethernet0", iface_id=100) + seed_metalbox_cache(interfaces=[(eth, False, ["10.0.0.1/24"])]) + device = SimpleNamespace(id=1, name="leaf-1") + + assert _get_metalbox_ip_for_device(device) == "10.0.0.1" + assert config_generator._metalbox_ip_cache[1] == "10.0.0.1" + + +def test_get_metalbox_ip_no_subnet_match_returns_none(mocker): + mocker.patch.object( + config_generator, "get_device_oob_ip", return_value=("10.0.0.5", 24) + ) + eth = make_iface("Ethernet0", iface_id=100) + seed_metalbox_cache(interfaces=[(eth, False, ["192.168.1.1/24"])]) + device = SimpleNamespace(id=1, name="leaf-1") + + assert _get_metalbox_ip_for_device(device) is None + assert config_generator._metalbox_ip_cache[1] is None + + +def test_get_metalbox_ip_skips_ipv6_addresses(mocker): + """``IPv4Address`` raises ``ValueError`` on an IPv6 string; the loop must + swallow it and keep looking at the next IP.""" + mocker.patch.object( + config_generator, "get_device_oob_ip", return_value=("10.0.0.5", 24) + ) + eth = make_iface("Ethernet0", iface_id=100) + seed_metalbox_cache( + interfaces=[ + (eth, False, ["2001:db8::1/64", "10.0.0.1/24"]), + ] + ) + device = SimpleNamespace(id=1, name="leaf-1") + + assert _get_metalbox_ip_for_device(device) == "10.0.0.1" + + +def test_get_metalbox_ip_returns_match_on_vlan_interface(mocker): + mocker.patch.object( + config_generator, "get_device_oob_ip", return_value=("10.0.0.5", 24) + ) + vlan = make_iface("Vlan100", type_value="virtual", iface_id=200) + seed_metalbox_cache(interfaces=[(vlan, True, ["10.0.0.1/24"])]) + device = SimpleNamespace(id=1, name="leaf-1") + + assert _get_metalbox_ip_for_device(device) == "10.0.0.1" + + +def test_get_metalbox_ip_second_call_hits_cache(mocker): + oob_mock = mocker.patch.object( + config_generator, "get_device_oob_ip", return_value=("10.0.0.5", 24) + ) + eth = make_iface("Ethernet0", iface_id=100) + seed_metalbox_cache(interfaces=[(eth, False, ["10.0.0.1/24"])]) + device = SimpleNamespace(id=1, name="leaf-1") + + first = _get_metalbox_ip_for_device(device) + second = _get_metalbox_ip_for_device(device) + + assert first == second == "10.0.0.1" + # Second call short-circuited on the IP cache; no extra OOB lookup. + oob_mock.assert_called_once_with(device) + + +def test_get_metalbox_ip_when_devices_cache_is_none(mocker): + mocker.patch.object( + config_generator, "get_device_oob_ip", return_value=("10.0.0.5", 24) + ) + # Autouse already left _metalbox_devices_cache at None. + assert config_generator._metalbox_devices_cache is None + device = SimpleNamespace(id=1, name="leaf-1") + + assert _get_metalbox_ip_for_device(device) is None + assert config_generator._metalbox_ip_cache[1] is None + + +def test_get_metalbox_ip_outer_exception_returns_none(mocker): + """A malformed OOB IP string causes ``IPv4Network`` to raise — the outer + ``try`` catches it and the device gets a ``None`` cached result.""" + mocker.patch.object( + config_generator, + "get_device_oob_ip", + return_value=("not-an-ip", 24), + ) + device = SimpleNamespace(id=1, name="leaf-1") + + assert _get_metalbox_ip_for_device(device) is None + assert config_generator._metalbox_ip_cache[1] is None diff --git a/tests/unit/tasks/conductor/sonic/test_config_generator_ntp.py b/tests/unit/tasks/conductor/sonic/test_config_generator_ntp.py new file mode 100644 index 00000000..8fcdfc47 --- /dev/null +++ b/tests/unit/tasks/conductor/sonic/test_config_generator_ntp.py @@ -0,0 +1,58 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for ``_add_ntp_configuration`` (per-device NTP wiring).""" + +from types import SimpleNamespace + +import pytest + +from osism.tasks.conductor.sonic import config_generator +from osism.tasks.conductor.sonic.config_generator import _add_ntp_configuration + +pytestmark = pytest.mark.usefixtures("reset_config_generator_caches") + + +# --------------------------------------------------------------------------- +# _add_ntp_configuration +# --------------------------------------------------------------------------- + + +def test_add_ntp_configuration_with_metalbox_ip(mocker): + mocker.patch.object( + config_generator, + "_get_metalbox_ip_for_device", + return_value="10.0.0.1", + ) + config = {"NTP_SERVER": {}} + + _add_ntp_configuration(config, SimpleNamespace(name="leaf-1")) + + assert config["NTP_SERVER"] == { + "10.0.0.1": {"maxpoll": "10", "minpoll": "6", "prefer": "false"} + } + + +def test_add_ntp_configuration_no_metalbox_ip_leaves_config_untouched(mocker): + mocker.patch.object( + config_generator, "_get_metalbox_ip_for_device", return_value=None + ) + config = {"NTP_SERVER": {}} + + _add_ntp_configuration(config, SimpleNamespace(name="leaf-1")) + + assert config["NTP_SERVER"] == {} + + +def test_add_ntp_configuration_helper_exception_swallowed(mocker): + """A failure inside the helper must never propagate out of the orchestrator — + NTP being unconfigured is a soft failure.""" + mocker.patch.object( + config_generator, + "_get_metalbox_ip_for_device", + side_effect=RuntimeError("kaboom"), + ) + config = {"NTP_SERVER": {}} + + _add_ntp_configuration(config, SimpleNamespace(name="leaf-1")) + + assert config["NTP_SERVER"] == {} diff --git a/tests/unit/tasks/conductor/sonic/test_config_generator_orchestrator.py b/tests/unit/tasks/conductor/sonic/test_config_generator_orchestrator.py new file mode 100644 index 00000000..d084e746 --- /dev/null +++ b/tests/unit/tasks/conductor/sonic/test_config_generator_orchestrator.py @@ -0,0 +1,556 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the SONiC ``config_generator`` orchestrator +(``generate_sonic_config``) and the public cache-clear helpers. + +Service-level behavior (metalbox cache, NTP, DNS, log-server, SNMP) lives in +sibling ``test_config_generator_.py`` files; this file deliberately +leaves their helpers patched so the orchestrator's own glue is exercised in +isolation. +""" + +from types import SimpleNamespace + +import pytest + +from osism.tasks.conductor.sonic import config_generator +from osism.tasks.conductor.sonic.config_generator import ( + TOP_LEVEL_SCAFFOLD_KEYS, + clear_all_caches, + clear_metalbox_devices_cache, + clear_metalbox_ip_cache, + generate_sonic_config, + natural_sort_key, +) + +from ._config_generator_helpers import make_base_config, patch_base_config + +pytestmark = pytest.mark.usefixtures("reset_config_generator_caches") + + +# --------------------------------------------------------------------------- +# Helpers / fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def patch_orchestrator_helpers(mocker): + """Patch every helper the orchestrator delegates to. + + Returns a ``SimpleNamespace`` exposing each mock so individual tests can + customize side effects (e.g. swap a ``return_value`` for a non-empty + ``oob_ip_result``). Each helper is patched at its **import site** in + ``config_generator`` — patching at the source module would not catch the + rebound reference the orchestrator already imported. + """ + + def patch(name, **kw): + return mocker.patch( + f"osism.tasks.conductor.sonic.config_generator.{name}", **kw + ) + + return SimpleNamespace( + get_port_config=patch("get_port_config", return_value={}), + detect_port_channels=patch( + "detect_port_channels", + return_value={"portchannels": {}, "member_mapping": {}}, + ), + detect_breakout_ports=patch( + "detect_breakout_ports", + return_value={"breakout_cfgs": {}, "breakout_ports": {}}, + ), + convert_netbox_interface_to_sonic=patch( + "convert_netbox_interface_to_sonic", + side_effect=lambda iface, _device: iface.name, + ), + get_speed_from_port_type=patch("get_speed_from_port_type", return_value=None), + get_connected_interfaces=patch( + "get_connected_interfaces", return_value=(set(), set()) + ), + get_device_oob_ip=patch("get_device_oob_ip", return_value=None), + get_device_vlans=patch("get_device_vlans", return_value={}), + get_device_loopbacks=patch("get_device_loopbacks", return_value={}), + get_device_interface_ips=patch("get_device_interface_ips", return_value={}), + get_device_platform=patch( + "get_device_platform", return_value="x86_64-generic-r0" + ), + get_device_hostname=patch("get_device_hostname", return_value="leaf-1"), + get_device_mac_address=patch( + "get_device_mac_address", return_value="aa:bb:cc:dd:ee:ff" + ), + _get_vrf_info=patch( + "_get_vrf_info", + return_value={"vrfs": {}, "interface_vrf_mapping": {}}, + ), + _get_transfer_role_ipv4_addresses=patch( + "_get_transfer_role_ipv4_addresses", return_value={} + ), + _add_port_configurations=patch("_add_port_configurations"), + _add_interface_configurations=patch("_add_interface_configurations"), + _add_bgp_configurations=patch("_add_bgp_configurations"), + _add_ntp_configuration=patch("_add_ntp_configuration"), + _add_dns_configuration=patch("_add_dns_configuration"), + _add_vlan_configuration=patch("_add_vlan_configuration"), + _add_loopback_configuration=patch("_add_loopback_configuration"), + _add_log_server_configuration=patch("_add_log_server_configuration"), + _add_snmp_configuration=patch("_add_snmp_configuration"), + _add_vrf_configuration=patch("_add_vrf_configuration"), + _add_portchannel_configuration=patch("_add_portchannel_configuration"), + get_cached_device_interfaces=patch( + "get_cached_device_interfaces", return_value=[] + ), + _get_metalbox_ip_for_device=patch( + "_get_metalbox_ip_for_device", return_value="10.0.0.1" + ), + calculate_local_asn_from_ipv4=patch( + "calculate_local_asn_from_ipv4", return_value=4200000001 + ), + ) + + +@pytest.fixture +def make_orchestrator_device(): + """Build a minimal NetBox-shaped device the orchestrator can consume.""" + + def _factory(device_id=1, name="leaf-1", primary_ip4=None, primary_ip6=None): + return SimpleNamespace( + id=device_id, + name=name, + primary_ip4=primary_ip4, + primary_ip6=primary_ip6, + ) + + return _factory + + +def _ip(address): + """Build a stub for ``device.primary_ip4`` / ``primary_ip6``.""" + return SimpleNamespace(address=address) + + +# --------------------------------------------------------------------------- +# natural_sort_key +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "port_name,expected", + [ + ("Ethernet0", 0), + ("Ethernet120", 120), + ("PortChannel", 0), # No digits → falls back to 0. + ("Ethernet1/3/2", 1), # Picks the first digit run. + ("", 0), # Empty string → no match → 0. + ], +) +def test_natural_sort_key(port_name, expected): + assert natural_sort_key(port_name) == expected + + +def test_natural_sort_key_returns_int(): + # Sorting routines depend on a comparable int — pin the type explicitly. + assert isinstance(natural_sort_key("Ethernet42"), int) + + +# --------------------------------------------------------------------------- +# generate_sonic_config — required keys / DEVICE_METADATA population +# --------------------------------------------------------------------------- + + +def test_generate_sonic_config_returns_required_top_level_keys( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker) + device = make_orchestrator_device() + + config = generate_sonic_config(device, "Test-HWSKU") + + # Every scaffold key the production code commits to must be present. + for key in TOP_LEVEL_SCAFFOLD_KEYS: + assert key in config + + +def test_generate_sonic_config_populates_device_metadata_localhost( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker) + patch_orchestrator_helpers.get_device_hostname.return_value = "spine-7" + patch_orchestrator_helpers.get_device_platform.return_value = "x86_64-foo-r0" + patch_orchestrator_helpers.get_device_mac_address.return_value = "11:22:33:44:55:66" + device = make_orchestrator_device(name="spine-7") + + config = generate_sonic_config(device, "Custom-HWSKU") + + assert config["DEVICE_METADATA"]["localhost"] == { + "hostname": "spine-7", + "hwsku": "Custom-HWSKU", + "platform": "x86_64-foo-r0", + "mac": "11:22:33:44:55:66", + } + + +def test_generate_sonic_config_preserves_existing_localhost_keys( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + """Pre-existing ``DEVICE_METADATA.localhost`` keys must survive the update. + + Production base configs ship platform-independent fields under + ``localhost`` (e.g. ``type=LeafRouter``); ``.update()`` should add / + overwrite the orchestrator's keys without dropping the rest. + """ + base = make_base_config() + base["DEVICE_METADATA"]["localhost"] = {"type": "LeafRouter", "hwsku": "OLD"} + patch_base_config(mocker, base_config=base) + device = make_orchestrator_device() + + config = generate_sonic_config(device, "NEW-HWSKU") + + assert config["DEVICE_METADATA"]["localhost"]["type"] == "LeafRouter" + assert config["DEVICE_METADATA"]["localhost"]["hwsku"] == "NEW-HWSKU" + + +# --------------------------------------------------------------------------- +# generate_sonic_config — base-config loading / deepcopy / fallback +# --------------------------------------------------------------------------- + + +def test_generate_sonic_config_uses_deepcopy_per_device( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + """Two devices must not share the same base-config dict object. + + Without ``deepcopy`` the second device's mutations would leak back into + the first device's config (and into the cached base config) — a real bug + we got bitten by before this test existed. + """ + base = make_base_config() + base["DEVICE_METADATA"]["shared"] = {"flag": "original"} + patch_base_config(mocker, base_config=base) + + device_a = make_orchestrator_device(device_id=1, name="dev-a") + device_b = make_orchestrator_device(device_id=2, name="dev-b") + + config_a = generate_sonic_config(device_a, "HWSKU-A") + # Mutate device_a's config; device_b must not see it. + config_a["DEVICE_METADATA"]["shared"]["flag"] = "mutated" + + config_b = generate_sonic_config(device_b, "HWSKU-B") + + assert config_b["DEVICE_METADATA"]["shared"] == {"flag": "original"} + assert config_a["DEVICE_METADATA"] is not config_b["DEVICE_METADATA"] + + +def test_generate_sonic_config_starts_from_scaffold_when_base_absent( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker, exists=False) + device = make_orchestrator_device() + + config = generate_sonic_config(device, "Test-HWSKU") + + # Scaffold keys exist even though ``open`` was never called. + for key in TOP_LEVEL_SCAFFOLD_KEYS: + assert key in config + assert config["DEVICE_METADATA"]["localhost"]["hwsku"] == "Test-HWSKU" + + +def test_generate_sonic_config_starts_fresh_when_base_load_raises( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker, exists=True, raise_on_open=OSError("disk on fire")) + device = make_orchestrator_device() + + config = generate_sonic_config(device, "Test-HWSKU") + + # Scaffold survived the failed load, hostname still made it through. + assert "DEVICE_METADATA" in config + assert config["DEVICE_METADATA"]["localhost"]["hwsku"] == "Test-HWSKU" + + +# --------------------------------------------------------------------------- +# generate_sonic_config — primary-IP / AS routing +# --------------------------------------------------------------------------- + + +def test_generate_sonic_config_router_id_and_local_asn_from_primary_ip4( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker) + patch_orchestrator_helpers.calculate_local_asn_from_ipv4.return_value = 4200045123 + device = make_orchestrator_device(primary_ip4=_ip("192.168.45.123/32")) + + config = generate_sonic_config(device, "Test-HWSKU") + + assert config["BGP_GLOBALS"]["default"]["router_id"] == "192.168.45.123" + assert config["BGP_GLOBALS"]["default"]["local_asn"] == "4200045123" + patch_orchestrator_helpers.calculate_local_asn_from_ipv4.assert_called_once_with( + "192.168.45.123" + ) + + +def test_generate_sonic_config_calculate_asn_value_error_logged_no_local_asn( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + """A ``ValueError`` from the AS calculator must be swallowed. + + ``router_id`` should still be set from ``primary_ip4`` and the function + must return successfully — only ``local_asn`` is omitted. + """ + patch_base_config(mocker) + patch_orchestrator_helpers.calculate_local_asn_from_ipv4.side_effect = ValueError( + "bad octet" + ) + device = make_orchestrator_device(primary_ip4=_ip("10.0.0.1/32")) + + config = generate_sonic_config(device, "HWSKU") + + assert config["BGP_GLOBALS"]["default"]["router_id"] == "10.0.0.1" + assert "local_asn" not in config["BGP_GLOBALS"]["default"] + + +def test_generate_sonic_config_router_id_falls_back_to_primary_ip6( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker) + device = make_orchestrator_device( + primary_ip4=None, primary_ip6=_ip("2001:db8::1/128") + ) + + config = generate_sonic_config(device, "HWSKU") + + assert config["BGP_GLOBALS"]["default"]["router_id"] == "2001:db8::1" + # IPv6-only path must not invoke the IPv4 ASN calculator. + patch_orchestrator_helpers.calculate_local_asn_from_ipv4.assert_not_called() + assert "local_asn" not in config["BGP_GLOBALS"].get("default", {}) + + +def test_generate_sonic_config_no_primary_ip_skips_bgp_globals_default( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + """Edge case: a device without any primary IP (rare in production but + worth pinning) must not produce a partial ``BGP_GLOBALS["default"]``.""" + patch_base_config(mocker) + device = make_orchestrator_device(primary_ip4=None, primary_ip6=None) + + config = generate_sonic_config(device, "HWSKU") + + assert "default" not in config["BGP_GLOBALS"] + patch_orchestrator_helpers.calculate_local_asn_from_ipv4.assert_not_called() + + +def test_generate_sonic_config_device_as_mapping_overrides_calculator( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + """Spine/superspine devices use a pre-computed group AS — the calculator + is bypassed entirely so the group's minimum AS wins.""" + patch_base_config(mocker) + device = make_orchestrator_device(device_id=42, primary_ip4=_ip("192.168.45.99/32")) + + config = generate_sonic_config(device, "HWSKU", device_as_mapping={42: 4200099999}) + + assert config["BGP_GLOBALS"]["default"]["local_asn"] == "4200099999" + patch_orchestrator_helpers.calculate_local_asn_from_ipv4.assert_not_called() + + +# --------------------------------------------------------------------------- +# generate_sonic_config — OOB / management routing +# --------------------------------------------------------------------------- + + +def test_generate_sonic_config_populates_mgmt_interface_and_static_route( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker) + patch_orchestrator_helpers.get_device_oob_ip.return_value = ("10.42.0.5", 24) + patch_orchestrator_helpers._get_metalbox_ip_for_device.return_value = "10.42.0.1" + device = make_orchestrator_device() + + config = generate_sonic_config(device, "HWSKU") + + assert config["MGMT_INTERFACE"]["eth0"] == {"admin_status": "up"} + assert "eth0|10.42.0.5/24" in config["MGMT_INTERFACE"] + assert config["STATIC_ROUTE"]["mgmt|0.0.0.0/0"] == {"nexthop": "10.42.0.1"} + patch_orchestrator_helpers._get_metalbox_ip_for_device.assert_called_once_with( + device + ) + # SNMP receives the OOB IP for SNMP_AGENT_ADDRESS_CONFIG wiring. + _, _, snmp_oob = patch_orchestrator_helpers._add_snmp_configuration.call_args.args + assert snmp_oob == "10.42.0.5" + + +def test_generate_sonic_config_no_oob_ip_leaves_mgmt_empty_and_passes_none( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker) + patch_orchestrator_helpers.get_device_oob_ip.return_value = None + device = make_orchestrator_device() + + config = generate_sonic_config(device, "HWSKU") + + assert config["MGMT_INTERFACE"] == {} + # STATIC_ROUTE was scaffolded but never populated. + assert "mgmt|0.0.0.0/0" not in config["STATIC_ROUTE"] + _, _, snmp_oob = patch_orchestrator_helpers._add_snmp_configuration.call_args.args + assert snmp_oob is None + + +# --------------------------------------------------------------------------- +# generate_sonic_config — breakout merge +# --------------------------------------------------------------------------- + + +def test_generate_sonic_config_merges_breakout_cfgs_and_ports( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker) + patch_orchestrator_helpers.detect_breakout_ports.return_value = { + "breakout_cfgs": {"Ethernet0": {"brkout_mode": "4x25G"}}, + "breakout_ports": {"Ethernet0": {"master": "Ethernet0"}}, + } + device = make_orchestrator_device() + + config = generate_sonic_config(device, "HWSKU") + + assert config["BREAKOUT_CFG"]["Ethernet0"] == {"brkout_mode": "4x25G"} + assert config["BREAKOUT_PORTS"]["Ethernet0"] == {"master": "Ethernet0"} + + +# --------------------------------------------------------------------------- +# generate_sonic_config — config_version normalization +# --------------------------------------------------------------------------- + + +def test_generate_sonic_config_version_short_form_gets_prefix( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker) + device = make_orchestrator_device() + + config = generate_sonic_config(device, "HWSKU", config_version="4_2_0") + + assert config["VERSIONS"]["DATABASE"]["VERSION"] == "version_4_2_0" + + +def test_generate_sonic_config_version_long_form_kept_as_is( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker) + device = make_orchestrator_device() + + config = generate_sonic_config(device, "HWSKU", config_version="version_4_2_0") + + assert config["VERSIONS"]["DATABASE"]["VERSION"] == "version_4_2_0" + + +def test_generate_sonic_config_version_default_when_missing( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker, base_config=make_base_config()) + device = make_orchestrator_device() + + config = generate_sonic_config(device, "HWSKU", config_version=None) + + assert config["VERSIONS"]["DATABASE"]["VERSION"] == "version_4_0_1" + + +def test_generate_sonic_config_version_existing_in_base_preserved( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + patch_base_config(mocker, base_config=make_base_config(version="version_4_5_0")) + device = make_orchestrator_device() + + config = generate_sonic_config(device, "HWSKU", config_version=None) + + assert config["VERSIONS"]["DATABASE"]["VERSION"] == "version_4_5_0" + + +# --------------------------------------------------------------------------- +# generate_sonic_config — netbox_interfaces collection +# --------------------------------------------------------------------------- + + +def test_generate_sonic_config_handles_interfaces_without_optional_fields( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + """``speed`` / ``type`` / ``tags`` are optional on NetBox interfaces — the + orchestrator must tolerate any of them being missing or ``None`` and fall + through to ``get_speed_from_port_type`` only when both are absent.""" + patch_base_config(mocker) + iface_no_speed = SimpleNamespace(name="Ethernet0", speed=None, type=None, tags=[]) + iface_with_type = SimpleNamespace( + name="Ethernet1", + speed=None, + type=SimpleNamespace(value="25gbase-x-sfp28"), + tags=[], + ) + patch_orchestrator_helpers.get_cached_device_interfaces.return_value = [ + iface_no_speed, + iface_with_type, + ] + patch_orchestrator_helpers.get_speed_from_port_type.return_value = 25_000_000 + + # Should not raise; the orchestrator builds netbox_interfaces internally + # before delegating to _add_port_configurations (which is mocked). + config = generate_sonic_config(make_orchestrator_device(), "HWSKU") + + assert isinstance(config, dict) + patch_orchestrator_helpers._add_port_configurations.assert_called_once() + + +def test_generate_sonic_config_get_cached_interfaces_failure_logged( + mocker, patch_orchestrator_helpers, make_orchestrator_device +): + """If the per-device interface fetch fails, ``netbox_interfaces`` stays + empty (warning logged) and the orchestrator continues — the section + helpers will simply receive an empty mapping.""" + patch_base_config(mocker) + patch_orchestrator_helpers.get_cached_device_interfaces.side_effect = RuntimeError( + "netbox unreachable" + ) + + config = generate_sonic_config(make_orchestrator_device(), "HWSKU") + + assert isinstance(config, dict) + # _add_port_configurations was still called with an (empty) dict. + args, _ = patch_orchestrator_helpers._add_port_configurations.call_args + # netbox_interfaces is the 6th positional argument: (config, port_config, + # connected_interfaces, portchannel_info, breakout_info, + # netbox_interfaces, vlan_info, device). + assert args[5] == {} + + +# --------------------------------------------------------------------------- +# clear_*_cache helpers +# --------------------------------------------------------------------------- + + +def test_clear_metalbox_ip_cache_resets_to_empty_dict(): + config_generator._metalbox_ip_cache = {1: "10.0.0.1"} + + clear_metalbox_ip_cache() + + assert config_generator._metalbox_ip_cache == {} + + +def test_clear_metalbox_devices_cache_resets_to_none(): + config_generator._metalbox_devices_cache = {1: {"device": object()}} + + clear_metalbox_devices_cache() + + assert config_generator._metalbox_devices_cache is None + + +def test_clear_all_caches_clears_everything_and_port_config(mocker): + """``clear_all_caches`` is the single entry point ``sync_sonic`` uses; + if any of the resets is dropped we get cross-run contamination.""" + cpc = mocker.patch( + "osism.tasks.conductor.sonic.config_generator.clear_port_config_cache" + ) + config_generator._metalbox_ip_cache = {1: "10.0.0.1"} + config_generator._metalbox_devices_cache = {1: {}} + + clear_all_caches() + + assert config_generator._metalbox_ip_cache == {} + assert config_generator._metalbox_devices_cache is None + cpc.assert_called_once_with() diff --git a/tests/unit/tasks/conductor/sonic/test_config_generator_snmp.py b/tests/unit/tasks/conductor/sonic/test_config_generator_snmp.py new file mode 100644 index 00000000..d11844b8 --- /dev/null +++ b/tests/unit/tasks/conductor/sonic/test_config_generator_snmp.py @@ -0,0 +1,171 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for ``_add_snmp_configuration`` in ``config_generator``.""" + +from types import SimpleNamespace + +import pytest + +from osism.tasks.conductor.sonic.config_generator import _add_snmp_configuration + +pytestmark = pytest.mark.usefixtures("reset_config_generator_caches") + + +@pytest.fixture +def patch_snmp_secrets(mocker): + """Patch ``deep_decrypt`` and ``get_vault`` to no-ops in the module's + namespace so the SNMP tests don't need a real vault. + + The orchestrator binds them at import time, so we patch the rebound + references on the ``config_generator`` module rather than on + ``conductor.utils``. + """ + + mocker.patch("osism.tasks.conductor.sonic.config_generator.get_vault") + mocker.patch( + "osism.tasks.conductor.sonic.config_generator.deep_decrypt", + side_effect=lambda data, vault: None, + ) + + +def _snmp_device(custom_fields=None, **ctx): + return SimpleNamespace( + name="leaf-1", + config_context=ctx, + custom_fields=custom_fields if custom_fields is not None else {}, + ) + + +def test_add_snmp_configuration_defaults(patch_snmp_secrets): + config = {} + + _add_snmp_configuration(config, _snmp_device(), oob_ip=None) + + assert config["SNMP_SERVER"]["SYSTEM"] == { + "sysContact": "info@example.com", + "sysLocation": "Data Center", + "traps": "enable", + } + + +def test_add_snmp_configuration_traps_disabled_omits_key(patch_snmp_secrets): + config = {} + + _add_snmp_configuration( + config, + _snmp_device(_segment_snmp_server_traps=False), + oob_ip=None, + ) + + assert "traps" not in config["SNMP_SERVER"]["SYSTEM"] + + +def test_add_snmp_configuration_oob_ip_populates_agent_address(patch_snmp_secrets): + config = {} + + _add_snmp_configuration(config, _snmp_device(), oob_ip="10.42.0.5") + + assert config["SNMP_AGENT_ADDRESS_CONFIG"] == { + "10.42.0.5|161|mgmt": {"name": "agentEntry1"} + } + + +def test_add_snmp_configuration_no_oob_ip_omits_agent_address(patch_snmp_secrets): + config = {} + + _add_snmp_configuration(config, _snmp_device(), oob_ip=None) + + assert "SNMP_AGENT_ADDRESS_CONFIG" not in config + + +def test_add_snmp_configuration_user_pulls_secrets_from_node_secrets( + patch_snmp_secrets, +): + config = {} + device = _snmp_device( + custom_fields={ + "secrets": { + "_segment_snmp_server_userauthpass": "auth-secret", + "_segment_snmp_server_userprivpass": "priv-secret", + } + }, + _segment_snmp_server_username="snmpuser", + ) + + _add_snmp_configuration(config, device, oob_ip=None) + + assert config["SNMP_SERVER_USER"]["snmpuser"] == { + "shaKey": "auth-secret", + "aesKey": "priv-secret", + } + assert config["SNMP_SERVER_GROUP_MEMBER"]["monitoring|snmpuser"] == { + "securityModel": ["usm"] + } + + +def test_add_snmp_configuration_user_falls_back_to_obfuscated_secrets( + patch_snmp_secrets, +): + config = {} + device = _snmp_device(_segment_snmp_server_username="snmpuser") + + _add_snmp_configuration(config, device, oob_ip=None) + + assert config["SNMP_SERVER_USER"]["snmpuser"] == { + "shaKey": "OBFUSCATEDAUTHSECRET", + "aesKey": "OBFUSCATEDPRIVSECRET", + } + + +def test_add_snmp_configuration_user_with_hosts_creates_targets(patch_snmp_secrets): + config = {} + device = _snmp_device( + _segment_snmp_server_username="snmpuser", + _segment_snmp_server_hosts=["10.0.0.10", "10.0.0.11"], + ) + + _add_snmp_configuration(config, device, oob_ip=None) + + assert set(config["SNMP_SERVER_PARAMS"].keys()) == { + "targetEntry1", + "targetEntry2", + } + assert config["SNMP_SERVER_TARGET"]["targetEntry1"]["ip"] == "10.0.0.10" + assert config["SNMP_SERVER_TARGET"]["targetEntry2"]["ip"] == "10.0.0.11" + # All targets must reference the user via SNMP_SERVER_PARAMS. + for entry in config["SNMP_SERVER_PARAMS"].values(): + assert entry == {"security-level": "auth-priv", "user": "snmpuser"} + + +def test_add_snmp_configuration_no_user_omits_user_target_sections( + patch_snmp_secrets, +): + config = {} + + _add_snmp_configuration(config, _snmp_device(), oob_ip=None) + + for absent in ( + "SNMP_SERVER_USER", + "SNMP_SERVER_GROUP_MEMBER", + "SNMP_SERVER_PARAMS", + "SNMP_SERVER_TARGET", + ): + assert absent not in config + + +def test_add_snmp_configuration_secrets_none_treated_as_empty(patch_snmp_secrets): + """``custom_fields["secrets"] is None`` is the production-default for + devices without an encrypted-secrets field — the helper must treat it as + ``{}`` rather than dereferencing ``None.get``.""" + config = {} + device = _snmp_device( + custom_fields={"secrets": None}, + _segment_snmp_server_username="snmpuser", + ) + + _add_snmp_configuration(config, device, oob_ip=None) + + assert config["SNMP_SERVER_USER"]["snmpuser"] == { + "shaKey": "OBFUSCATEDAUTHSECRET", + "aesKey": "OBFUSCATEDPRIVSECRET", + }