Fix napalm_rpc_map override, netmiko error, and napalm_formula arg bugs

changelog/69797.fixed.md

@@ -0,0 +1,7 @@
+Fixed four bugs in the ``napalm_mod`` and ``napalm_formula`` execution modules.
+``napalm.rpc`` now honours a user-supplied ``napalm_rpc_map`` override instead of
+letting the built-in defaults clobber it; ``napalm.netmiko_args`` raises a clear
+error (rather than a raw ``KeyError``) for an ``os`` grain with no Netmiko device
+type; ``napalm_formula.container_path`` now honours its ``key``/``container``/``delim``
+arguments; and ``napalm_formula.render_field`` no longer raises ``KeyError`` when the
+``os`` grain is absent.

salt/modules/napalm_formula.py

@@ -90,7 +90,7 @@ def container_path(model, key=None, container=None, delim=DEFAULT_TARGET_DELIM):
         - interfaces:interface:Ethernet1:subinterfaces:subinterface:0:config
         - interfaces:interface:Ethernet2:config
     """
-    return list(_container_path(model))
+    return list(_container_path(model, key=key, container=container, delim=delim))
 
 
 def setval(key, val, dict_=None, delim=DEFAULT_TARGET_DELIM):
@@ -287,7 +287,7 @@ def render_field(dictionary, field, prepend=None, append=None, quotes=False, **o
     if prepend is None:
         prepend = field.replace("_", "-")
     if append is None:
-        if __grains__["os"] in ("junos",):
+        if __grains__.get("os") in ("junos",):
             append = ";"
         else:
             append = ""

salt/modules/napalm_mod.py

@@ -529,7 +529,14 @@ def netmiko_args(**kwargs):
     netmiko_device_type_map.update(
         __salt__["config.get"]("netmiko_device_type_map", {})
     )
-    kwargs["device_type"] = netmiko_device_type_map[__grains__["os"]]
+    os_grain = __grains__.get("os")
+    if os_grain not in netmiko_device_type_map:
+        raise CommandExecutionError(
+            "Unable to map the '{}' NAPALM driver to a Netmiko device type. "
+            "Please add it to the netmiko_device_type_map configuration option "
+            "/ Pillar.".format(os_grain)
+        )
+    kwargs["device_type"] = netmiko_device_type_map[os_grain]
     return kwargs
 
 
@@ -1406,8 +1413,11 @@ def rpc(command, **kwargs):
         "eos": "napalm.pyeapi_run_commands",
         "nxos": "napalm.nxos_api_rpc",
     }
-    napalm_map = __salt__["config.get"]("napalm_rpc_map", {})
-    napalm_map.update(default_map)
+    # User-supplied napalm_rpc_map entries must override the built-in defaults
+    # (the old order let default_map win), and we must not mutate the object
+    # config.get returns; start from the defaults and layer the user map on top.
+    napalm_map = dict(default_map)
+    napalm_map.update(__salt__["config.get"]("napalm_rpc_map", {}))
     fun = napalm_map.get(__grains__["os"], "napalm.netmiko_commands")
     return __salt__[fun](command, **kwargs)
 

tests/pytests/unit/modules/napalm/test_formula.py

@@ -196,3 +196,20 @@ def test_render_fields():
     )
     ret = napalm_formula.render_fields(config, "mtu", "description", quotes=True)
     assert ret == expected_render
+
+
+def test_container_path_uses_delim(set_model):
+    # Regression: container_path dropped its delim (and key/container), always
+    # using the default ':'. With delim='//' no ':' should appear in the paths.
+    with patch("salt.utils.napalm.is_proxy", MagicMock(return_value=True)):
+        ret = napalm_formula.container_path(set_model.copy(), delim="//")
+    assert "interfaces//interface//Ethernet1//config" in ret
+    assert not any(":" in path for path in ret)
+
+
+def test_render_field_no_os_grain():
+    # 'os' grain absent must not raise KeyError; no junos trailing ';'.
+    config = {"description": "Interface description"}
+    with patch.dict(napalm_formula.__grains__, {}, clear=True):
+        ret = napalm_formula.render_field(config, "description", quotes=True)
+    assert ret == 'description "Interface description"'

tests/pytests/unit/modules/napalm/test_mod.py

@@ -8,6 +8,7 @@
 
 import salt.modules.napalm_mod as napalm_mod
 import tests.support.napalm as napalm_test_support
+from salt.exceptions import CommandExecutionError
 from tests.support.mock import MagicMock, patch
 
 log = logging.getLogger(__file__)
@@ -206,3 +207,50 @@ def test_config_kwargs_werid_transport_port():
         ret = napalm_mod.pyeapi_nxos_api_args(kwargs=test_kwargs)
         assert ret["transport"] == "nxos_protocol"
         assert ret["port"] == 2080
+
+
+def test_rpc_user_map_overrides_default():
+    # A user-supplied napalm_rpc_map entry must win over the built-in default
+    # (the old order let default_map clobber it), without mutating the config.
+    user_map = {"junos": "napalm.custom_rpc"}
+    custom = MagicMock(return_value="custom-result")
+    with patch.dict(
+        napalm_mod.__salt__,
+        {
+            "config.get": MagicMock(return_value=user_map),
+            "napalm.custom_rpc": custom,
+        },
+    ), patch.dict(napalm_mod.__grains__, {"os": "junos"}):
+        # Call the undecorated body; the proxy_napalm_wrap decorator would try to
+        # open a real device (this fix is in the function body, not the wrapper).
+        ret = napalm_mod.rpc.__wrapped__("show version")
+    custom.assert_called_once_with("show version")
+    assert ret == "custom-result"
+    # the config object returned by config.get must not be mutated with defaults
+    assert user_map == {"junos": "napalm.custom_rpc"}
+
+
+def test_netmiko_args_unknown_os_raises_clean_error():
+    # An os grain not in the map (custom/community driver, no user override)
+    # must raise a clear CommandExecutionError, not a raw KeyError.
+    napalm_opts = {
+        "HOSTNAME": "device",
+        "USERNAME": "user",
+        "PASSWORD": "pass",
+        "TIMEOUT": 60,
+        "OPTIONAL_ARGS": {},
+    }
+    with patch(
+        "salt.utils.napalm.get_device_opts", MagicMock(return_value=napalm_opts)
+    ), patch.object(napalm_mod, "HAS_NETMIKO", True), patch.object(
+        napalm_mod, "_get_netmiko_args", MagicMock(return_value={})
+    ), patch.dict(
+        napalm_mod.__salt__, {"config.get": MagicMock(return_value={})}
+    ), patch.dict(
+        napalm_mod.__grains__, {"os": "customdriver"}
+    ):
+        with pytest.raises(CommandExecutionError) as exc:
+            napalm_mod.netmiko_args.__wrapped__()
+    # Specifically the "no device type for this driver" error (naming the os),
+    # not the earlier "netmiko is not installed" gate.
+    assert "customdriver" in str(exc.value)