Add unit tests for SONiC config_generator orchestrator and service ca…

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()

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}

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."""

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"] == {}

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